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

feat(triggers): nested filter groups and dotted paths (#10625)

Browse Source 
* feat(triggers): nested any_of / all_of filter groups

A trigger filter entry can now be a group — `{"any_of": [...]}` or
`{"all_of": [...]}` — nesting further entries, so criteria like
`A AND B AND (C OR D)` are expressible. Existing flat `{key, value}` lists keep
their meaning, combined by the trigger's `filter_logic` as before.

Filters are compiled once per connection: the set of top-level keys the whole
tree references is collected up front, so a message is parsed in a single
streaming pass that captures only those keys, instead of one full pass per leaf
filter as before. Filters that fail to parse are now logged rather than dropped
silently, since a nested group is easier to mistype than a flat entry.

The editor gains "Add group", rendering groups recursively with their own
AND/OR selector; Kafka and WebSocket triggers share it.

Fixes WIN-2345

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(triggers): drop empty filter groups instead of evaluating them

A group with no criterion cannot evaluate to a constant: true makes an `or`
filter list accept every message, false mutes an `and` list. Two clicks in the
editor ("Add group", save) produced one. Drop it when compiling so its siblings
stay in force, and reject at save time the filters the listener would otherwise
drop silently.

Also restore the item shape of `$ref`-typed arrays in the generated agent
schemas: the extractor only resolved refs at the property level, so moving
`filters.items` to a shared schema flattened it to a bare object. Resolving them
inside `items` too also recovers the shapes `initial_messages` and the MQTT
`topics` had already lost.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(triggers): name the offending entry when a nested filter is invalid

Serde's untagged error only reports that the outermost entry matched no
variant, whatever depth is actually wrong, which defeats the point of
validating a group at save time. Walk the tree instead and report the path.

Normalize the WebSocket editor's filters to [] on load, as the Kafka editor
does, so the list component can rely on an array.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(triggers): key filter rows by node so deletion keeps values aligned

The value editor seeds itself from `code` once, so an index-keyed row reused
for a different filter kept showing the deleted row's value.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* chore: bump ee-repo-ref after merging main

The merge pulled OSS code that needs EE symbols newer than the companion
branch's base, so the companion was merged with EE main too.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* perf(triggers): keep filter short-circuiting from materializing unread fields

The single-pass scan deserialized every referenced key before the boolean tree
ran, so an AND whose first leaf rejects the message still allocated the large
objects the later leaves name — the shape this feature exists for. Borrow the
wanted keys as raw slices during the scan and parse a field only when
evaluation actually reaches it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* feat(triggers): none_of filter group

Negation of a nested group, so a trigger can exclude what it must not react to
without inverting every other criterion. A key the message does not carry
satisfies it: there is nothing there to match.

Only groups can negate — the root's operator is the trigger's filter_logic
column, which has no value for it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* feat(triggers): address a nested field with a dotted path

`{path: "a.b.c", value: v}` alongside the existing `{key, value}`, so the common
case reads the way people write it instead of nesting the shape into the value.
A separate field rather than dots in `key`, which already means the top-level
field spelled that way — overloading it would resettle what existing triggers
over flattened payloads match.

Paths address objects only for now: a path through an array does not match
rather than guessing an element, and array containment stays on the value side.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* docs(triggers): mention none_of in the filter_logic description

Plus a test for the empty-path-segment rejection, which had none.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* chore: update ee-repo-ref to 78859aab0c6e78283ec8d2b37e8c410963afdc83

This commit updates the EE repository reference after PR #722 was merged in windmill-ee-private.

Previous ee-repo-ref: 0e42ba72ccc38a6b0a380f58afe0db36d284f4c9

New ee-repo-ref: 78859aab0c6e78283ec8d2b37e8c410963afdc83

Automated by sync-ee-ref workflow.

* fix(triggers): reject a criterion naming both key and path

The untagged enum takes such an entry as a `key` criterion and drops the
`path`, which is the silent-ignore the save-time validation exists to prevent.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* refactor(triggers): drop the label next to the key/path toggle

The toggle already shows which one is selected.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(triggers): reject an entry that combines a criterion with a group

Generalizes the key+path fix: the untagged enum settles a half-and-half entry
on the first variant that fits and ignores the rest, so a criterion carrying a
group key lost the whole subtree without a word.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
This commit is contained in:
Ruben Fiszel
2026-08-11 11:07:12 +02:00
committed by GitHub
parent 5125467de4
commit ec99108cf6
17 changed files with 1280 additions and 293 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
04dd9c5c352f04995cd0470400a877261f956561
78859aab0c6e78283ec8d2b37e8c410963afdc83
backend/windmill-api/openapi.yaml
+68 -57
View File
@@ -29112,6 +29112,56 @@ components:
- interval_secs
- message
TriggerFilter:
description: >
Either a leaf filter, matching a field of the message (parsed as JSON) against a
value by equality (or superset, when the value is an object or array) — addressed
by `key` for a top-level field or `path` for a dotted path into nested objects —
or a group nesting sub-filters under a boolean operator (`none_of` matches when
none of its sub-filters do).
oneOf:
- type: object
properties:
key:
type: string
value: {}
required:
- key
- value
- type: object
properties:
path:
type: string
description: Dotted path into nested objects, e.g. `a.b.c`. Does not traverse arrays.
value: {}
required:
- path
- value
- type: object
properties:
any_of:
type: array
items:
$ref: "#/components/schemas/TriggerFilter"
required:
- any_of
- type: object
properties:
all_of:
type: array
items:
$ref: "#/components/schemas/TriggerFilter"
required:
- all_of
- type: object
properties:
none_of:
type: array
items:
$ref: "#/components/schemas/TriggerFilter"
required:
- none_of
WebsocketTrigger:
allOf:
- $ref: "#/components/schemas/TriggerExtraProperty"
@@ -29132,23 +29182,16 @@ components:
description: Last error message if the trigger failed
filters:
type: array
description: Array of key-value filters to match incoming messages (only matching messages trigger the script)
description: "Filters to match incoming messages (only matching messages trigger the script). Each entry is either a leaf `{key, value}` (top-level field) or `{path, value}` (dotted path into nested objects), or a group `{any_of: [...]}` / `{all_of: [...]}` / `{none_of: [...]}` nesting more entries. Entries at the top level are combined with `filter_logic`."
items:
type: object
properties:
key:
type: string
value: {}
required:
- key
- value
$ref: "#/components/schemas/TriggerFilter"
filter_logic:
type: string
enum:
- and
- or
default: and
description: "Logic to apply when evaluating filters. 'and' requires all filters to match, 'or' requires any filter to match."
description: "Logic to apply when evaluating the top-level filters. 'and' requires all of them to match, 'or' requires any of them to match. Nested `any_of`/`all_of`/`none_of` groups carry their own logic."
initial_messages:
type: array
nullable: true
@@ -29204,23 +29247,16 @@ components:
$ref: "#/components/schemas/TriggerMode"
filters:
type: array
description: Array of key-value filters to match incoming messages (only matching messages trigger the script)
description: "Filters to match incoming messages (only matching messages trigger the script). Each entry is either a leaf `{key, value}` (top-level field) or `{path, value}` (dotted path into nested objects), or a group `{any_of: [...]}` / `{all_of: [...]}` / `{none_of: [...]}` nesting more entries. Entries at the top level are combined with `filter_logic`."
items:
type: object
properties:
key:
type: string
value: {}
required:
- key
- value
$ref: "#/components/schemas/TriggerFilter"
filter_logic:
type: string
enum:
- and
- or
default: and
description: "Logic to apply when evaluating filters. 'and' requires all filters to match, 'or' requires any filter to match."
description: "Logic to apply when evaluating the top-level filters. 'and' requires all of them to match, 'or' requires any of them to match. Nested `any_of`/`all_of`/`none_of` groups carry their own logic."
initial_messages:
type: array
nullable: true
@@ -29287,23 +29323,16 @@ components:
description: True if script_path points to a flow, false if it points to a script
filters:
type: array
description: Array of key-value filters to match incoming messages (only matching messages trigger the script)
description: "Filters to match incoming messages (only matching messages trigger the script). Each entry is either a leaf `{key, value}` (top-level field) or `{path, value}` (dotted path into nested objects), or a group `{any_of: [...]}` / `{all_of: [...]}` / `{none_of: [...]}` nesting more entries. Entries at the top level are combined with `filter_logic`."
items:
type: object
properties:
key:
type: string
value: {}
required:
- key
- value
$ref: "#/components/schemas/TriggerFilter"
filter_logic:
type: string
enum:
- and
- or
default: and
description: "Logic to apply when evaluating filters. 'and' requires all filters to match, 'or' requires any filter to match."
description: "Logic to apply when evaluating the top-level filters. 'and' requires all of them to match, 'or' requires any of them to match. Nested `any_of`/`all_of`/`none_of` groups carry their own logic."
initial_messages:
type: array
nullable: true
@@ -30559,22 +30588,16 @@ components:
description: Array of Kafka topic names to subscribe to
filters:
type: array
description: "Filters to match incoming messages (only matching messages trigger the script). Each entry is either a leaf `{key, value}` (top-level field) or `{path, value}` (dotted path into nested objects), or a group `{any_of: [...]}` / `{all_of: [...]}` / `{none_of: [...]}` nesting more entries. Entries at the top level are combined with `filter_logic`."
items:
type: object
properties:
key:
type: string
value: {}
required:
- key
- value
$ref: "#/components/schemas/TriggerFilter"
filter_logic:
type: string
enum:
- and
- or
default: and
description: "Logic to apply when evaluating filters. 'and' requires all filters to match, 'or' requires any filter to match."
description: "Logic to apply when evaluating the top-level filters. 'and' requires all of them to match, 'or' requires any of them to match. Nested `any_of`/`all_of`/`none_of` groups carry their own logic."
auto_offset_reset:
type: string
enum:
@@ -30637,22 +30660,16 @@ components:
description: Array of Kafka topic names to subscribe to
filters:
type: array
description: "Filters to match incoming messages (only matching messages trigger the script). Each entry is either a leaf `{key, value}` (top-level field) or `{path, value}` (dotted path into nested objects), or a group `{any_of: [...]}` / `{all_of: [...]}` / `{none_of: [...]}` nesting more entries. Entries at the top level are combined with `filter_logic`."
items:
type: object
properties:
key:
type: string
value: {}
required:
- key
- value
$ref: "#/components/schemas/TriggerFilter"
filter_logic:
type: string
enum:
- and
- or
default: and
description: "Logic to apply when evaluating filters. 'and' requires all filters to match, 'or' requires any filter to match."
description: "Logic to apply when evaluating the top-level filters. 'and' requires all of them to match, 'or' requires any of them to match. Nested `any_of`/`all_of`/`none_of` groups carry their own logic."
auto_offset_reset:
type: string
enum:
@@ -30711,22 +30728,16 @@ components:
description: Array of Kafka topic names to subscribe to
filters:
type: array
description: "Filters to match incoming messages (only matching messages trigger the script). Each entry is either a leaf `{key, value}` (top-level field) or `{path, value}` (dotted path into nested objects), or a group `{any_of: [...]}` / `{all_of: [...]}` / `{none_of: [...]}` nesting more entries. Entries at the top level are combined with `filter_logic`."
items:
type: object
properties:
key:
type: string
value: {}
required:
- key
- value
$ref: "#/components/schemas/TriggerFilter"
filter_logic:
type: string
enum:
- and
- or
default: and
description: "Logic to apply when evaluating filters. 'and' requires all filters to match, 'or' requires any filter to match."
description: "Logic to apply when evaluating the top-level filters. 'and' requires all of them to match, 'or' requires any of them to match. Nested `any_of`/`all_of`/`none_of` groups carry their own logic."
auto_offset_reset:
type: string
enum:
backend/windmill-trigger-websocket/src/handler.rs
+3 -1
View File
@@ -12,7 +12,7 @@ use windmill_common::{
worker::to_raw_value,
};
use windmill_git_sync::DeployedObject;
use windmill_trigger::{Trigger, TriggerCrud, TriggerData};
use windmill_trigger::{filter::CompiledFilters, Trigger, TriggerCrud, TriggerData};
use super::{
get_url_from_runnable_value, listener::InitialMessage, proxy::connect_async_with_proxy,
@@ -106,6 +106,8 @@ impl TriggerCrud for WebsocketTrigger {
}
}
CompiledFilters::validate(&config.filters)?;
if let Some(ref hb) = config.heartbeat {
if hb.interval_secs < 1 {
return Err(Error::BadRequest(
backend/windmill-trigger-websocket/src/listener.rs
+9 -12
View File
@@ -21,7 +21,7 @@ use windmill_common::{
DB,
};
use windmill_queue::PushArgsOwned;
use windmill_trigger::filter::{check_filters, Filter};
use windmill_trigger::filter::CompiledFilters;
use windmill_trigger::listener::{update_rw_lock, ListeningTrigger};
use windmill_trigger::trigger_helpers::{
trigger_runnable, trigger_runnable_and_wait_for_raw_result,
@@ -362,15 +362,14 @@ impl Listener for WebsocketTrigger {
} => {},
// Message reader
_ = async {
let filters: Vec<Filter> = if listening_trigger.trigger_mode {
listening_trigger
.trigger_config
.filters
.iter()
.filter_map(|m| serde_json::from_str(m.get()).ok())
.collect_vec()
let filters = if listening_trigger.trigger_mode {
CompiledFilters::parse(
listening_trigger.trigger_config.filters.iter().map(|m| m.get()),
listening_trigger.trigger_config.filter_logic == "or",
&listening_trigger.path,
)
} else {
vec![]
CompiledFilters::default()
};
loop {
if let Some(msg) = reader.next().await {
@@ -391,9 +390,7 @@ impl Listener for WebsocketTrigger {
}
}
let use_or = listening_trigger.trigger_config.filter_logic == "or";
let should_handle = check_filters(&text, &filters, use_or);
if should_handle {
if filters.matches(&text) {
let trigger_info = HashMap::from([
("url".to_string(), to_raw_value(&listening_trigger.trigger_config.url)),
]);
backend/windmill-trigger/src/filter.rs
+593 -81
View File
@@ -2,52 +2,301 @@ use serde::{
de::{self, MapAccess, Visitor},
Deserialize, Deserializer,
};
use serde_json::Value;
use std::fmt;
use serde_json::{value::RawValue, Value};
use std::{collections::HashMap, fmt};
#[derive(Deserialize)]
#[derive(Debug, Deserialize)]
pub struct JsonFilter {
pub key: String,
pub value: Value,
}
#[derive(Deserialize)]
/// Same comparison as [`JsonFilter`], but the field is addressed by a dotted path into
/// nested objects. A separate field rather than dots in `key`, because a `key` containing
/// a dot already means the top-level field spelled that way.
#[derive(Debug, Deserialize)]
pub struct PathFilter {
pub path: String,
pub value: Value,
}
/// Boolean group of nested filters, externally tagged (`{"any_of": [...]}`) so it is
/// unambiguous against a leaf filter, which is `{"key": ..., "value": ...}`.
#[derive(Debug, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum FilterGroup {
AnyOf(Vec<Filter>),
AllOf(Vec<Filter>),
NoneOf(Vec<Filter>),
}
#[derive(Debug, Deserialize)]
#[serde(untagged)]
pub enum Filter {
JsonFilter(JsonFilter),
PathFilter(PathFilter),
Group(FilterGroup),
}
struct SupersetVisitor<'a> {
key: &'a str,
value_to_check: &'a Value,
/// The scanned top-level key, and whatever is left to walk inside its value.
fn split_path(path: &str) -> (&str, &str) {
path.split_once('.').unwrap_or((path, ""))
}
impl<'de, 'a> Visitor<'de> for SupersetVisitor<'a> {
type Value = bool;
/// Objects only: `Value::get` yields nothing for a string index into an array, so a path
/// through one simply does not match rather than guessing an element.
fn resolve<'v>(root: &'v Value, rest: &str) -> Option<&'v Value> {
let mut current = root;
if !rest.is_empty() {
for segment in rest.split('.') {
current = current.get(segment)?;
}
}
Some(current)
}
/// Filters prepared for repeated evaluation against a stream of messages. The set of
/// top-level keys the whole tree references is computed once, so each message is scanned
/// in a single pass instead of once per leaf filter.
#[derive(Debug, Default)]
pub struct CompiledFilters {
filters: Vec<Filter>,
use_or_logic: bool,
keys: Vec<String>,
}
impl CompiledFilters {
pub fn new(filters: Vec<Filter>, use_or_logic: bool) -> Self {
let filters = drop_empty_groups(filters);
let mut keys = Vec::new();
collect_keys(&filters, &mut keys);
Self { filters, use_or_logic, keys }
}
/// Build from the raw JSON of each filter as stored in the trigger config. An entry
/// that fails to parse is skipped rather than dropping the other filters, but it
/// widens what the trigger accepts, so it is reported.
pub fn parse<'a>(
raw_filters: impl IntoIterator<Item = &'a str>,
use_or_logic: bool,
trigger_path: &str,
) -> Self {
let filters = raw_filters
.into_iter()
.filter_map(|raw| match serde_json::from_str::<Filter>(raw) {
Ok(filter) => Some(filter),
Err(err) => {
tracing::error!(
"Ignoring unparseable filter of trigger {}: {} ({})",
trigger_path,
raw,
err
);
None
}
})
.collect();
Self::new(filters, use_or_logic)
}
pub fn is_empty(&self) -> bool {
self.filters.is_empty()
}
/// Reject at save time what [`Self::parse`] would drop at listen time. A group nests
/// arbitrarily many criteria, so one mistyped entry silently widens the trigger by the
/// whole subtree it belongs to.
pub fn validate(filters: &[Value]) -> windmill_common::error::Result<()> {
for (index, filter) in filters.iter().enumerate() {
validate_filter(filter, &format!("filter #{}", index + 1))?;
}
Ok(())
}
/// Whether `text`, parsed as a JSON object, satisfies the filters.
pub fn matches(&self, text: &str) -> bool {
if self.filters.is_empty() {
return true;
}
let mut deserializer = serde_json::Deserializer::from_str(text);
let values =
Deserializer::deserialize_map(&mut deserializer, KeysVisitor { keys: &self.keys })
.unwrap_or_default();
eval_all(&self.filters, self.use_or_logic, &values)
}
}
/// Groups are descended into by hand so a bad entry is named on its own: serde's untagged
/// error only reports that the outermost entry matched no variant, whatever depth is wrong.
fn validate_filter(filter: &Value, path: &str) -> windmill_common::error::Result<()> {
const GROUP_KEYS: [&str; 3] = ["any_of", "all_of", "none_of"];
let group = filter
.as_object()
.filter(|object| object.len() == 1)
.and_then(|object| {
GROUP_KEYS
.into_iter()
.find_map(|key| object.get(key).map(|nested| (key, nested)))
});
if let Some((key, nested)) = group {
let nested = nested.as_array().ok_or_else(|| {
windmill_common::error::Error::BadRequest(format!(
"{}: {} must be an array of filters",
path, key
))
})?;
for (index, child) in nested.iter().enumerate() {
validate_filter(child, &format!("{} -> {}[{}]", path, key, index))?;
}
return Ok(());
}
// Everything below is meant to be a leaf. The untagged enum resolves a half-and-half
// entry by taking the first variant that fits and ignoring the rest of it, so a
// criterion carrying a group key would silently lose the whole subtree.
if let Some(group_key) = GROUP_KEYS.iter().find(|key| filter.get(*key).is_some()) {
return Err(windmill_common::error::Error::BadRequest(format!(
"{} combines a criterion with a {} group; an entry is one or the other",
path, group_key
)));
}
if filter.get("key").is_some() && filter.get("path").is_some() {
return Err(windmill_common::error::Error::BadRequest(format!(
"{} names its field with both key and path; use one or the other",
path
)));
}
let parsed = serde_json::from_value::<Filter>(filter.clone()).map_err(|err| {
windmill_common::error::Error::BadRequest(format!(
"{} is neither a {{key, value}} / {{path, value}} criterion nor an any_of/all_of/none_of group: {}",
path, err
))
})?;
// An empty segment addresses no field, so the filter could only ever reject everything
if let Filter::PathFilter(PathFilter { path: dotted, .. }) = &parsed {
if dotted.split('.').any(|segment| segment.is_empty()) {
return Err(windmill_common::error::Error::BadRequest(format!(
"{}: path {:?} has an empty segment",
path, dotted
)));
}
}
Ok(())
}
/// A group with no criterion cannot evaluate to a constant: `true` makes an `or` list
/// accept every message, `false` mutes an `and` list. Dropping it instead leaves its
/// siblings in force, which is what a group left empty in the editor should mean.
fn drop_empty_groups(filters: Vec<Filter>) -> Vec<Filter> {
filters
.into_iter()
.filter_map(|filter| {
let (rebuild, nested): (fn(Vec<Filter>) -> FilterGroup, _) = match filter {
Filter::Group(FilterGroup::AnyOf(nested)) => (FilterGroup::AnyOf, nested),
Filter::Group(FilterGroup::AllOf(nested)) => (FilterGroup::AllOf, nested),
Filter::Group(FilterGroup::NoneOf(nested)) => (FilterGroup::NoneOf, nested),
leaf => return Some(leaf),
};
let nested = drop_empty_groups(nested);
(!nested.is_empty()).then(|| Filter::Group(rebuild(nested)))
})
.collect()
}
fn push_key(keys: &mut Vec<String>, key: &str) {
if !keys.iter().any(|k| k == key) {
keys.push(key.to_string());
}
}
fn collect_keys(filters: &[Filter], keys: &mut Vec<String>) {
for filter in filters {
match filter {
Filter::JsonFilter(JsonFilter { key, .. }) => push_key(keys, key),
Filter::PathFilter(PathFilter { path, .. }) => push_key(keys, split_path(path).0),
Filter::Group(
FilterGroup::AnyOf(nested)
| FilterGroup::AllOf(nested)
| FilterGroup::NoneOf(nested),
) => collect_keys(nested, keys),
}
}
}
/// `filters` is never empty: the top level is short-circuited by [`CompiledFilters::matches`],
/// and [`drop_empty_groups`] removes empty groups.
fn eval_all(filters: &[Filter], use_or_logic: bool, values: &HashMap<&str, &RawValue>) -> bool {
let eval = |filter: &Filter| match filter {
// Parsed here rather than during the scan so that `any`/`all` short-circuiting keeps
// a large field the verdict never depends on from being materialized at all.
Filter::JsonFilter(JsonFilter { key, value }) => values
.get(key.as_str())
.and_then(|raw| serde_json::from_str::<Value>(raw.get()).ok())
.map_or(false, |found| is_superset(&found, value)),
Filter::PathFilter(PathFilter { path, value }) => {
let (root, rest) = split_path(path);
values
.get(root)
.and_then(|raw| serde_json::from_str::<Value>(raw.get()).ok())
.and_then(|found| resolve(&found, rest).map(|at| is_superset(at, value)))
.unwrap_or(false)
}
Filter::Group(FilterGroup::AnyOf(nested)) => eval_all(nested, true, values),
Filter::Group(FilterGroup::AllOf(nested)) => eval_all(nested, false, values),
// A key the message does not carry satisfies a negation: nothing there can match.
Filter::Group(FilterGroup::NoneOf(nested)) => !eval_all(nested, true, values),
};
if use_or_logic {
filters.iter().any(eval)
} else {
filters.iter().all(eval)
}
}
/// Locates the requested top-level keys in a single pass, skipping every other value. The
/// ones it wants are borrowed as raw slices of the message rather than deserialized, so a
/// key the boolean evaluation never reaches costs nothing beyond the scan.
struct KeysVisitor<'k> {
keys: &'k [String],
}
impl<'de, 'k> Visitor<'de> for KeysVisitor<'k> {
type Value = HashMap<&'k str, &'de RawValue>;
fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
formatter.write_str("a JSON object with a specific key at the top level")
formatter.write_str("a JSON object")
}
fn visit_map<V>(self, mut map: V) -> std::result::Result<Self::Value, V::Error>
where
V: MapAccess<'de>,
{
let mut result = false;
let mut found = false;
let mut found = HashMap::with_capacity(self.keys.len());
// Must consume entire map to satisfy deserializer contract
while let Some(key) = map.next_key::<String>()? {
if !found && key == self.key {
let json_value: Value = map.next_value()?;
result = is_superset(&json_value, self.value_to_check);
found = true;
} else {
// Skip values we don't need (cheaper than full deserialization)
let _ = map.next_value::<de::IgnoredAny>()?;
match self.keys.iter().find(|k| k.as_str() == key) {
// On a duplicated key the first occurrence wins
Some(k) if !found.contains_key(k.as_str()) => {
found.insert(k.as_str(), map.next_value::<&'de RawValue>()?);
}
_ => {
// Skip values we don't need (cheaper than full deserialization)
let _ = map.next_value::<de::IgnoredAny>()?;
}
}
}
Ok(result)
Ok(found)
}
}
@@ -69,96 +318,359 @@ pub fn is_superset(json_value: &Value, value_to_check: &Value) -> bool {
}
}
pub fn is_value_superset<'a, 'de, D>(
deserializer: D,
key: &'a str,
value_to_check: &'a Value,
) -> std::result::Result<bool, D::Error>
where
D: Deserializer<'de>,
{
deserializer.deserialize_map(SupersetVisitor { key, value_to_check })
}
pub fn check_filters(text: &str, filters: &[Filter], use_or_logic: bool) -> bool {
if filters.is_empty() {
return true;
}
let check = |filter: &Filter| -> bool {
match filter {
Filter::JsonFilter(JsonFilter { key, value }) => {
let mut deserializer = serde_json::Deserializer::from_str(text);
is_value_superset(&mut deserializer, key, value).unwrap_or(false)
}
}
};
if use_or_logic {
filters.iter().any(check)
} else {
filters.iter().all(check)
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
fn matches(payload: &str, filters: serde_json::Value, use_or_logic: bool) -> bool {
let filters: Vec<Filter> = serde_json::from_value(filters).unwrap();
CompiledFilters::new(filters, use_or_logic).matches(payload)
}
#[test]
fn test_filter_with_other_top_level_keys() {
let payload = r#"{"event_type": "test", "other": "data"}"#;
let key = "event_type";
let value = json!("test");
let mut deserializer = serde_json::Deserializer::from_str(payload);
let result = is_value_superset(&mut deserializer, key, &value).unwrap();
assert!(result, "Should match when key exists with correct value");
let filters = json!([{"key": "event_type", "value": "test"}]);
assert!(
matches(payload, filters, false),
"Should match when key exists with correct value"
);
}
#[test]
fn test_filter_with_key_not_first() {
let payload = r#"{"other": "data", "event_type": "test"}"#;
let key = "event_type";
let value = json!("test");
let mut deserializer = serde_json::Deserializer::from_str(payload);
let result = is_value_superset(&mut deserializer, key, &value).unwrap();
assert!(result, "Should match even when key is not first");
let filters = json!([{"key": "event_type", "value": "test"}]);
assert!(
matches(payload, filters, false),
"Should match even when key is not first"
);
}
#[test]
fn test_filter_with_nested_object() {
let payload = r#"{"data": {"status": "active", "count": 5}, "other": "value"}"#;
let key = "data";
let value = json!({"status": "active"});
let mut deserializer = serde_json::Deserializer::from_str(payload);
let result = is_value_superset(&mut deserializer, key, &value).unwrap();
assert!(result, "Should match when nested object is superset");
let filters = json!([{"key": "data", "value": {"status": "active"}}]);
assert!(
matches(payload, filters, false),
"Should match when nested object is superset"
);
}
#[test]
fn test_filter_no_match() {
let payload = r#"{"event_type": "other", "data": "value"}"#;
let key = "event_type";
let value = json!("test");
let mut deserializer = serde_json::Deserializer::from_str(payload);
let result = is_value_superset(&mut deserializer, key, &value).unwrap();
assert!(!result, "Should not match when value differs");
let filters = json!([{"key": "event_type", "value": "test"}]);
assert!(
!matches(payload, filters, false),
"Should not match when value differs"
);
}
#[test]
fn test_filter_key_not_found() {
let payload = r#"{"other": "data"}"#;
let key = "event_type";
let value = json!("test");
let filters = json!([{"key": "event_type", "value": "test"}]);
assert!(
!matches(payload, filters, false),
"Should not match when key doesn't exist"
);
}
let mut deserializer = serde_json::Deserializer::from_str(payload);
let result = is_value_superset(&mut deserializer, key, &value).unwrap();
assert!(!result, "Should not match when key doesn't exist");
#[test]
fn test_no_filters_matches_everything() {
assert!(matches(r#"{"a": 1}"#, json!([]), false));
assert!(matches("not even json", json!([]), true));
}
#[test]
fn test_non_object_payload_never_matches() {
let filters = json!([{"key": "a", "value": 1}]);
assert!(!matches("[1, 2]", filters.clone(), false));
assert!(!matches("nope", filters, true));
}
#[test]
fn test_top_level_and_or_logic() {
let payload = r#"{"a": 1, "b": 2}"#;
let filters = json!([{"key": "a", "value": 1}, {"key": "b", "value": 99}]);
assert!(!matches(payload, filters.clone(), false));
assert!(matches(payload, filters, true));
}
// --- nested groups ---
#[test]
fn test_any_of_group_nested_in_and() {
let payload =
r#"{"event": "message_created", "previous_message": {"sent_by": "reminder"}}"#;
let filters = json!([
{"key": "event", "value": "message_created"},
{"any_of": [
{"key": "in_reply_to", "value": {"sent_by": "reminder"}},
{"key": "previous_message", "value": {"sent_by": "reminder"}}
]}
]);
assert!(matches(payload, filters, false));
}
#[test]
fn test_any_of_group_all_branches_fail() {
let payload = r#"{"event": "message_created", "previous_message": {"sent_by": "someone"}}"#;
let filters = json!([
{"key": "event", "value": "message_created"},
{"any_of": [
{"key": "in_reply_to", "value": {"sent_by": "reminder"}},
{"key": "previous_message", "value": {"sent_by": "reminder"}}
]}
]);
assert!(!matches(payload, filters, false));
}
#[test]
fn test_all_of_group_nested_in_or() {
let payload = r#"{"a": 1, "b": 2}"#;
let filters = json!([
{"key": "missing", "value": true},
{"all_of": [{"key": "a", "value": 1}, {"key": "b", "value": 2}]}
]);
assert!(matches(payload, filters.clone(), true));
assert!(!matches(payload, filters, false));
}
/// `1e400` overflows `Value`'s f64 and only fails to parse if something reads it, so a
/// match here means the short-circuit really did skip that field rather than
/// materializing every referenced key up front.
#[test]
fn test_unreached_branch_is_never_materialized() {
let payload = r#"{"gate": "match", "huge": 1e400}"#;
let filters = json!([{"key": "gate", "value": "match"}, {"key": "huge", "value": 1}]);
assert!(matches(payload, filters.clone(), true));
assert!(!matches(payload, filters, false));
}
#[test]
fn test_deeply_nested_groups() {
let payload = r#"{"a": 1, "b": 2, "c": 3}"#;
let filters = json!([
{"any_of": [
{"key": "a", "value": 99},
{"all_of": [
{"key": "b", "value": 2},
{"any_of": [{"key": "c", "value": 3}, {"key": "c", "value": 4}]}
]}
]}
]);
assert!(matches(payload, filters, false));
}
#[test]
fn test_path_reaches_a_nested_field() {
let payload = r#"{"in_reply_to_message": {"content_attributes": {"sent_by": "reminder"}}}"#;
let path = "in_reply_to_message.content_attributes.sent_by";
assert!(matches(
payload,
json!([{"path": path, "value": "reminder"}]),
false
));
assert!(!matches(
payload,
json!([{"path": path, "value": "other"}]),
false
));
// a missing intermediate segment is a miss, not an error
assert!(!matches(
payload,
json!([{"path": "in_reply_to_message.nope.sent_by", "value": "reminder"}]),
false
));
}
#[test]
fn test_path_and_key_keep_their_own_meaning() {
// `key` still addresses the top-level field spelled with dots, `path` traverses
assert!(matches(
r#"{"a.b": 1}"#,
json!([{"key": "a.b", "value": 1}]),
false
));
assert!(!matches(
r#"{"a.b": 1}"#,
json!([{"path": "a.b", "value": 1}]),
false
));
assert!(matches(
r#"{"a": {"b": 1}}"#,
json!([{"path": "a.b", "value": 1}]),
false
));
assert!(!matches(
r#"{"a": {"b": 1}}"#,
json!([{"key": "a.b", "value": 1}]),
false
));
}
#[test]
fn test_path_does_not_traverse_arrays() {
// Deliberately unsupported for now: an element index is not implied
assert!(!matches(
r#"{"items": [{"id": 1}]}"#,
json!([{"path": "items.id", "value": 1}]),
false
));
}
#[test]
fn test_path_without_dots_is_a_top_level_field() {
assert!(matches(
r#"{"a": 1}"#,
json!([{"path": "a", "value": 1}]),
false
));
}
#[test]
fn test_none_of_excludes_matching_messages() {
let filters = json!([
{"key": "event", "value": "message_created"},
{"none_of": [{"key": "sender", "value": "bot"}, {"key": "kind", "value": "draft"}]}
]);
assert!(matches(
r#"{"event": "message_created", "sender": "human"}"#,
filters.clone(),
false
));
assert!(!matches(
r#"{"event": "message_created", "sender": "bot"}"#,
filters.clone(),
false
));
// any one branch matching is enough to exclude
assert!(!matches(
r#"{"event": "message_created", "sender": "human", "kind": "draft"}"#,
filters,
false
));
}
#[test]
fn test_none_of_is_satisfied_by_a_missing_key() {
// Nothing is there to match, so the negation holds — the alternative would make
// every negative filter also require the field to be present.
assert!(matches(
r#"{"event": "message_created"}"#,
json!([{"none_of": [{"key": "sender", "value": "bot"}]}]),
false
));
}
#[test]
fn test_empty_group_is_dropped_not_constant() {
let payload = r#"{"a": 1}"#;
// On its own it leaves the trigger unfiltered, like an empty filter list
assert!(matches(payload, json!([{"any_of": []}]), false));
assert!(matches(
payload,
json!([{"all_of": []}, {"any_of": [{"all_of": []}]}]),
true
));
// Alongside a real criterion it must not decide the outcome either way
let with_failing_leaf = json!([{"any_of": []}, {"key": "a", "value": 99}]);
assert!(!matches(payload, with_failing_leaf.clone(), true));
assert!(!matches(payload, with_failing_leaf, false));
}
#[test]
fn test_parses_legacy_and_group_entries_side_by_side() {
let filters = CompiledFilters::parse(
[
r#"{"key": "event", "value": "created"}"#,
r#"{"any_of": [{"key": "a", "value": 1}, {"key": "b", "value": 2}]}"#,
],
false,
"u/admin/trigger",
);
assert!(filters.matches(r#"{"event": "created", "b": 2}"#));
assert!(!filters.matches(r#"{"event": "created", "b": 3}"#));
assert!(!filters.matches(r#"{"event": "other", "a": 1}"#));
}
#[test]
fn test_duplicated_payload_key_resolves_to_first_occurrence() {
let filters = json!([{"key": "a", "value": 1}]);
assert!(matches(r#"{"a": 1, "a": 2}"#, filters.clone(), false));
assert!(!matches(r#"{"a": 2, "a": 1}"#, filters, false));
}
#[test]
fn test_validate_rejects_entries_the_listener_would_drop() {
assert!(CompiledFilters::validate(&[
json!({"key": "a", "value": 1}),
json!({"all_of": []})
])
.is_ok());
assert!(
CompiledFilters::validate(&[json!({"anyOf": [{"key": "a", "value": 1}]})]).is_err()
);
assert!(CompiledFilters::validate(&[json!({"key": "a"})]).is_err());
}
#[test]
fn test_validate_rejects_a_leaf_that_also_carries_a_group() {
// Untagged would settle each of these on one variant and drop the rest of the entry
for mixed in [
json!({"key": "a", "value": 1, "none_of": [{"key": "b", "value": 2}]}),
json!({"path": "a.b", "value": 1, "any_of": [{"key": "b", "value": 2}]}),
json!({"any_of": [{"key": "a", "value": 1}], "all_of": [{"key": "b", "value": 2}]}),
] {
let err = CompiledFilters::validate(&[mixed.clone()])
.unwrap_err()
.to_string();
assert!(
err.contains("combines a criterion with"),
"{} should be rejected, got: {}",
mixed,
err
);
}
}
#[test]
fn test_validate_rejects_a_leaf_naming_both_key_and_path() {
// Untagged would take it as a `key` criterion and drop the `path` without a word
let err = CompiledFilters::validate(&[json!({"key": "a", "path": "b.c", "value": 1})])
.unwrap_err()
.to_string();
assert!(err.contains("both key and path"), "got: {}", err);
}
#[test]
fn test_validate_rejects_an_empty_path_segment() {
assert!(CompiledFilters::validate(&[json!({"path": "a.b", "value": 1})]).is_ok());
for dead in ["", "a.", ".a", "a..b"] {
assert!(
CompiledFilters::validate(&[json!({"path": dead, "value": 1})]).is_err(),
"path {:?} addresses no field and should be rejected",
dead
);
}
}
#[test]
fn test_validate_names_the_offending_nested_entry() {
let err = CompiledFilters::validate(&[
json!({"key": "a", "value": 1}),
json!({"all_of": [{"key": "b", "value": 2}, {"any_of": [{"key": "c"}]}]}),
])
.unwrap_err()
.to_string();
assert!(
err.contains("filter #2 -> all_of[1] -> any_of[0]"),
"error should point at the entry that is wrong, got: {}",
err
);
}
// --- is_superset unit tests ---
cli/src/guidance/skills.gen.ts
Generated
+140 -17
View File
@@ -8287,18 +8287,62 @@ properties:
filters:
type: array
items:
type: object
properties:
key:
type: string
value: {}
oneOf:
- type: object
properties:
key:
type: string
value: {}
required:
- key
- value
- type: object
properties:
path:
type: string
description: Dotted path into nested objects, e.g. \`a.b.c\`. Does not traverse
arrays.
value: {}
required:
- path
- value
- type: object
properties:
any_of:
type: array
items:
type: object
required:
- any_of
- type: object
properties:
all_of:
type: array
items:
type: object
required:
- all_of
- type: object
properties:
none_of:
type: array
items:
type: object
required:
- none_of
description: 'Filters to match incoming messages (only matching messages trigger
the script). Each entry is either a leaf \`{key, value}\` (top-level field) or
\`{path, value}\` (dotted path into nested objects), or a group \`{any_of: [...]}\`
/ \`{all_of: [...]}\` / \`{none_of: [...]}\` nesting more entries. Entries at the
top level are combined with \`filter_logic\`.'
filter_logic:
type: string
enum:
- and
- or
description: Logic to apply when evaluating filters. 'and' requires all filters
to match, 'or' requires any filter to match.
description: Logic to apply when evaluating the top-level filters. 'and' requires
all of them to match, 'or' requires any of them to match. Nested \`any_of\`/\`all_of\`/\`none_of\`
groups carry their own logic.
auto_offset_reset:
type: string
enum:
@@ -8399,6 +8443,18 @@ properties:
type: array
items:
type: object
properties:
qos:
type: string
enum:
- qos0
- qos1
- qos2
topic:
type: string
required:
- qos
- topic
description: Array of MQTT topics to subscribe to, each with topic name and QoS
level
v3_config:
@@ -8943,24 +8999,91 @@ properties:
filters:
type: array
items:
type: object
properties:
key:
type: string
value: {}
description: Array of key-value filters to match incoming messages (only matching
messages trigger the script)
oneOf:
- type: object
properties:
key:
type: string
value: {}
required:
- key
- value
- type: object
properties:
path:
type: string
description: Dotted path into nested objects, e.g. \`a.b.c\`. Does not traverse
arrays.
value: {}
required:
- path
- value
- type: object
properties:
any_of:
type: array
items:
type: object
required:
- any_of
- type: object
properties:
all_of:
type: array
items:
type: object
required:
- all_of
- type: object
properties:
none_of:
type: array
items:
type: object
required:
- none_of
description: 'Filters to match incoming messages (only matching messages trigger
the script). Each entry is either a leaf \`{key, value}\` (top-level field) or
\`{path, value}\` (dotted path into nested objects), or a group \`{any_of: [...]}\`
/ \`{all_of: [...]}\` / \`{none_of: [...]}\` nesting more entries. Entries at the
top level are combined with \`filter_logic\`.'
filter_logic:
type: string
enum:
- and
- or
description: Logic to apply when evaluating filters. 'and' requires all filters
to match, 'or' requires any filter to match.
description: Logic to apply when evaluating the top-level filters. 'and' requires
all of them to match, 'or' requires any of them to match. Nested \`any_of\`/\`all_of\`/\`none_of\`
groups carry their own logic.
initial_messages:
type: array
items:
type: object
oneOf:
- type: object
properties:
raw_message:
type: string
required:
- raw_message
- type: object
properties:
runnable_result:
type: object
properties:
path:
type: string
args:
type: object
description: The arguments to pass to the script or flow
additionalProperties: true
is_flow:
type: boolean
required:
- path
- args
- is_flow
required:
- runnable_result
description: Messages to send immediately after connecting (can be raw strings
or computed by runnables)
url_runnable_args:
frontend/src/lib/components/copilot/chat/workspaceToolsZod.gen.ts
Generated
+24 -6
View File
@@ -97,11 +97,20 @@ export const websocketTriggerRequestSchema = z.object({
"is_flow": z.boolean().describe("True if script_path points to a flow, false if it points to a script"),
"url": z.string().describe("The WebSocket URL to connect to (can be a static URL or computed by a runnable)"),
"mode": z.enum(["enabled", "disabled", "suspended"]).describe("job trigger mode").optional(),
"filters": z.array(z.object({
"filters": z.array(z.union([z.object({
"key": z.string(),
"value": z.any()
})).describe("Array of key-value filters to match incoming messages (only matching messages trigger the script)"),
"filter_logic": z.enum(["and", "or"]).describe("Logic to apply when evaluating filters. 'and' requires all filters to match, 'or' requires any filter to match.").default("and").optional(),
}), z.object({
"path": z.string().describe("Dotted path into nested objects, e.g. `a.b.c`. Does not traverse arrays."),
"value": z.any()
}), z.object({
"any_of": z.array(z.record(z.string(), z.any()))
}), z.object({
"all_of": z.array(z.record(z.string(), z.any()))
}), z.object({
"none_of": z.array(z.record(z.string(), z.any()))
})]).describe("Either a leaf filter, matching a field of the message (parsed as JSON) against a value by equality (or superset, when the value is an object or array) \u2014 addressed by `key` for a top-level field or `path` for a dotted path into nested objects \u2014 or a group nesting sub-filters under a boolean operator (`none_of` matches when none of its sub-filters do).\n")).describe("Filters to match incoming messages (only matching messages trigger the script). Each entry is either a leaf `{key, value}` (top-level field) or `{path, value}` (dotted path into nested objects), or a group `{any_of: [...]}` / `{all_of: [...]}` / `{none_of: [...]}` nesting more entries. Entries at the top level are combined with `filter_logic`."),
"filter_logic": z.enum(["and", "or"]).describe("Logic to apply when evaluating the top-level filters. 'and' requires all of them to match, 'or' requires any of them to match. Nested `any_of`/`all_of`/`none_of` groups carry their own logic.").default("and").optional(),
"initial_messages": z.array(z.union([z.object({
"raw_message": z.string()
}), z.object({
@@ -148,11 +157,20 @@ export const kafkaTriggerRequestSchema = z.object({
"kafka_resource_path": z.string().describe("Path to the Kafka resource containing connection configuration"),
"group_id": z.string().describe("Kafka consumer group ID for this trigger"),
"topics": z.array(z.string()).describe("Array of Kafka topic names to subscribe to"),
"filters": z.array(z.object({
"filters": z.array(z.union([z.object({
"key": z.string(),
"value": z.any()
})),
"filter_logic": z.enum(["and", "or"]).describe("Logic to apply when evaluating filters. 'and' requires all filters to match, 'or' requires any filter to match.").default("and").optional(),
}), z.object({
"path": z.string().describe("Dotted path into nested objects, e.g. `a.b.c`. Does not traverse arrays."),
"value": z.any()
}), z.object({
"any_of": z.array(z.record(z.string(), z.any()))
}), z.object({
"all_of": z.array(z.record(z.string(), z.any()))
}), z.object({
"none_of": z.array(z.record(z.string(), z.any()))
})]).describe("Either a leaf filter, matching a field of the message (parsed as JSON) against a value by equality (or superset, when the value is an object or array) \u2014 addressed by `key` for a top-level field or `path` for a dotted path into nested objects \u2014 or a group nesting sub-filters under a boolean operator (`none_of` matches when none of its sub-filters do).\n")).describe("Filters to match incoming messages (only matching messages trigger the script). Each entry is either a leaf `{key, value}` (top-level field) or `{path, value}` (dotted path into nested objects), or a group `{any_of: [...]}` / `{all_of: [...]}` / `{none_of: [...]}` nesting more entries. Entries at the top level are combined with `filter_logic`."),
"filter_logic": z.enum(["and", "or"]).describe("Logic to apply when evaluating the top-level filters. 'and' requires all of them to match, 'or' requires any of them to match. Nested `any_of`/`all_of`/`none_of` groups carry their own logic.").default("and").optional(),
"auto_offset_reset": z.enum(["latest", "earliest"]).describe("Initial offset behavior when consumer group has no committed offset.").default("latest").optional(),
"auto_commit": z.boolean().describe("When true (default), offsets are committed automatically after receiving each message. When false, you must manually commit offsets using the commit_offsets endpoint.").default(true).optional(),
"mode": z.enum(["enabled", "disabled", "suspended"]).describe("job trigger mode").optional(),
frontend/src/lib/components/triggers/TriggerFilterList.svelte
+161
View File
@@ -0,0 +1,161 @@
<script lang="ts">
import { Button } from '$lib/components/common'
import TextInput from '$lib/components/text_input/TextInput.svelte'
import Select from '$lib/components/select/Select.svelte'
import JsonEditor from '$lib/components/JsonEditor.svelte'
import { Plus, X } from 'lucide-svelte'
import { fade } from 'svelte/transition'
import TriggerFilterList from './TriggerFilterList.svelte'
import ToggleButtonGroup from '$lib/components/common/toggleButton-v2/ToggleButtonGroup.svelte'
import ToggleButton from '$lib/components/common/toggleButton-v2/ToggleButton.svelte'
import {
fieldMode,
groupItems,
groupOp,
isFilterGroup,
leafField,
makeGroup,
makeLeaf,
type GroupOp,
type FilterNode
} from './filters'
interface Props {
filters: FilterNode[]
logic: GroupOp
disabled?: boolean
/** Nesting level, 0 at the top. */
depth?: number
}
let { filters = $bindable(), logic = $bindable(), disabled = false, depth = 0 }: Props = $props()
// Deeper nesting is supported by the backend but stops being readable in this editor.
const MAX_DEPTH = 3
// Negation is offered on groups only: the root's operator is the trigger's
// `filter_logic` column, which has no value for it.
let logicItems = $derived([
{ label: 'all criteria (AND)', value: 'and' as const },
{ label: 'any criterion (OR)', value: 'or' as const },
...(depth > 0 ? [{ label: 'no criterion (NONE)', value: 'none' as const }] : [])
])
function add(node: FilterNode) {
filters = [...filters, node]
}
</script>
<div class="flex flex-col gap-2">
{#if depth > 0 || filters.length > 0}
<div class="max-w-xs">
<Select items={logicItems} bind:value={logic} {disabled} size="sm" />
</div>
{/if}
<!-- Keyed by node, not index: the value editor seeds itself from `code` once, so reusing
a row for a different filter after a deletion would leave the old value on screen. -->
{#each filters as filter, i (filter)}
<div class="flex w-full gap-2 items-start">
{#if isFilterGroup(filter)}
<div class="w-full border p-2 rounded-md bg-surface-secondary">
<TriggerFilterList
bind:filters={
() => groupItems(filter), (items) => (filters[i] = makeGroup(groupOp(filter), items))
}
bind:logic={
() => groupOp(filter),
(nested) => (filters[i] = makeGroup(nested, groupItems(filter)))
}
{disabled}
depth={depth + 1}
/>
</div>
{:else}
{@const mode = fieldMode(filter)}
{@const field = leafField(filter)}
<div class="w-full flex flex-col gap-2 border p-2 rounded-md bg-surface">
<div class="flex flex-col w-full">
<div class="flex flex-row items-center mb-2">
<ToggleButtonGroup
selected={mode}
{disabled}
on:selected={(e) => (filters[i] = makeLeaf(e.detail, field, filter.value))}
>
{#snippet children({ item })}
<ToggleButton
value="key"
label="Key"
small
{item}
tooltip="A top-level field of the message"
/>
<ToggleButton
value="path"
label="Path"
small
{item}
tooltip="A dotted path into nested objects, e.g. a.b.c. Does not traverse arrays."
/>
{/snippet}
</ToggleButtonGroup>
</div>
{#if 'path' in filter}
<TextInput bind:value={filter.path} inputProps={{ disabled, placeholder: 'a.b.c' }} />
{:else}
<TextInput bind:value={filter.key} inputProps={{ disabled }} />
{/if}
</div>
<div class="flex flex-col w-full">
<div class="text-secondary text-sm mb-2">Value</div>
<JsonEditor bind:value={filter.value} code={JSON.stringify(filter.value)} {disabled} />
</div>
{#if field}
{@const isObject = filter.value !== null && typeof filter.value === 'object'}
<div class="text-xs text-tertiary font-mono mt-2 p-2 bg-surface-secondary rounded">
payload.{field}
{isObject ? '⊇' : '=='}
{JSON.stringify(filter.value)}
</div>
{/if}
</div>
{/if}
<button
transition:fade|local={{ duration: 100 }}
class="rounded-full p-1 bg-surface-secondary duration-200 hover:bg-surface-hover mt-1"
aria-label="Clear"
onclick={() => {
filters = filters.filter((_, index) => index !== i)
}}
{disabled}
>
<X size={14} />
</button>
</div>
{/each}
<div class="flex items-baseline gap-2">
<Button
variant="default"
size="xs"
btnClasses="mt-1"
onclick={() => add({ key: '', value: '' })}
{disabled}
startIcon={{ icon: Plus }}
>
Add filter
</Button>
{#if depth < MAX_DEPTH}
<Button
variant="default"
size="xs"
btnClasses="mt-1"
onclick={() => add(makeGroup(logic === 'or' ? 'and' : 'or', []))}
{disabled}
startIcon={{ icon: Plus }}
>
Add group
</Button>
{/if}
</div>
</div>
frontend/src/lib/components/triggers/TriggerFilters.svelte
+14 -74
View File
@@ -1,14 +1,11 @@
<script lang="ts">
import { Button } from '$lib/components/common'
import Section from '$lib/components/Section.svelte'
import Select from '$lib/components/select/Select.svelte'
import { Plus, X } from 'lucide-svelte'
import { fade } from 'svelte/transition'
import JsonEditor from '$lib/components/JsonEditor.svelte'
import TriggerFilterList from './TriggerFilterList.svelte'
import type { FilterLogic, FilterNode, GroupOp } from './filters'
interface Props {
filters: { key: string; value: any }[]
filterLogic: 'and' | 'or'
filters: FilterNode[]
filterLogic: FilterLogic
disabled?: boolean
// Set when the runnable receives the payload base64-encoded (e.g. Kafka).
// Filters always run on the message parsed as JSON, so we clarify the distinction.
@@ -16,16 +13,16 @@
}
let {
filters = $bindable([]),
filters = $bindable(),
filterLogic = $bindable(),
disabled = false,
payloadBase64Encoded = false
}: Props = $props()
const filterLogicItems = [
{ label: 'all criteria (AND)', value: 'and' as const },
{ label: 'any criterion (OR)', value: 'or' as const }
]
// Only groups can negate, so the list never hands the root a 'none' back.
function setRootLogic(op: GroupOp) {
if (op !== 'none') filterLogic = op
}
let description = $derived(
filterLogic === 'or'
@@ -34,8 +31,9 @@
)
let filterHelp = $derived(
'The JSON filter checks if the value at the key is equal or a superset of the filter value. ' +
'Keys match top-level fields of the message (parsed as JSON); to match a nested field, set an object value (e.g. key data, value {"status": "active"}).' +
'Each criterion checks that the field is equal to, or a superset of, the filter value. ' +
'A Key names a top-level field of the message (parsed as JSON); a Path reaches a nested one, e.g. data.status. Paths do not traverse arrays — match those with an array value instead. ' +
'Add a group to nest criteria under their own logic, e.g. an OR of two fields inside an AND, or a NONE group to exclude messages that match it.' +
(payloadBase64Encoded
? ' The runnable still receives the payload base64-encoded; filters run on the message before that encoding.'
: '')
@@ -47,65 +45,7 @@
{description}<br />
{filterHelp}
</p>
{#if filters.length > 0}
<div class="mt-2 mb-1 max-w-xs">
<Select items={filterLogicItems} bind:value={filterLogic} {disabled} size="sm" />
</div>
{/if}
<div class="flex flex-col gap-4 mt-1">
{#each filters as v, i (i)}
<div class="flex w-full gap-2 items-center">
<div class="w-full flex flex-col gap-2 border p-2 rounded-md">
<label class="flex flex-col w-full">
<div class="text-secondary text-sm mb-2">Key</div>
<input type="text" bind:value={v.key} {disabled} />
</label>
<div class="flex flex-col w-full">
<div class="text-secondary text-sm mb-2">Value</div>
<JsonEditor bind:value={v.value} code={JSON.stringify(v.value)} {disabled} />
</div>
{#if v.key}
{@const isObject = v.value !== null && typeof v.value === 'object'}
<div class="text-xs text-tertiary font-mono mt-2 p-2 bg-surface-secondary rounded">
payload.{v.key}
{isObject ? '⊇' : '=='}
{JSON.stringify(v.value)}
</div>
{/if}
</div>
<button
transition:fade|local={{ duration: 100 }}
class="rounded-full p-1 bg-surface-secondary duration-200 hover:bg-surface-hover"
aria-label="Clear"
onclick={() => {
filters = filters.filter((_, index) => index !== i)
}}
{disabled}
>
<X size={14} />
</button>
</div>
{/each}
<div class="flex items-baseline">
<Button
variant="default"
size="xs"
btnClasses="mt-1"
onclick={() => {
if (filters == undefined || !Array.isArray(filters)) {
filters = []
}
filters = filters.concat({
key: '',
value: ''
})
}}
{disabled}
startIcon={{ icon: Plus }}
>
Add filter
</Button>
</div>
<div class="mt-1">
<TriggerFilterList bind:filters bind:logic={() => filterLogic, setRootLogic} {disabled} />
</div>
</Section>
frontend/src/lib/components/triggers/filters.ts
+58
View File
@@ -0,0 +1,58 @@
/** The root's operator, stored separately as the trigger's `filter_logic`. */
export type FilterLogic = 'and' | 'or'
/** A nested group's operator. Only groups can negate; the root has nowhere to store it. */
export type GroupOp = FilterLogic | 'none'
/** How a leaf addresses its field: a top-level name, or a dotted path into nested objects. */
export type FieldMode = 'key' | 'path'
export type FilterKeyLeaf = { key: string; value: any }
export type FilterPathLeaf = { path: string; value: any }
export type FilterLeaf = FilterKeyLeaf | FilterPathLeaf
export type FilterAnyOf = { any_of: FilterNode[] }
export type FilterAllOf = { all_of: FilterNode[] }
export type FilterNoneOf = { none_of: FilterNode[] }
export type FilterGroup = FilterAnyOf | FilterAllOf | FilterNoneOf
export type FilterNode = FilterLeaf | FilterGroup
export function isFilterGroup(node: FilterNode): node is FilterGroup {
return (
node != null &&
(Array.isArray((node as FilterAnyOf).any_of) ||
Array.isArray((node as FilterAllOf).all_of) ||
Array.isArray((node as FilterNoneOf).none_of))
)
}
export function groupOp(group: FilterGroup): GroupOp {
if ('any_of' in group) return 'or'
if ('none_of' in group) return 'none'
return 'and'
}
export function groupItems(group: FilterGroup): FilterNode[] {
if ('any_of' in group) return group.any_of
if ('none_of' in group) return group.none_of
return group.all_of
}
export function fieldMode(leaf: FilterLeaf): FieldMode {
return typeof (leaf as FilterPathLeaf).path === 'string' ? 'path' : 'key'
}
export function leafField(leaf: FilterLeaf): string {
return (
(fieldMode(leaf) === 'path' ? (leaf as FilterPathLeaf).path : (leaf as FilterKeyLeaf).key) ?? ''
)
}
/** A leaf names its field under one key or the other, so switching rebuilds the object. */
export function makeLeaf(mode: FieldMode, field: string, value: any): FilterLeaf {
return mode === 'path' ? { path: field, value } : { key: field, value }
}
/** Groups carry their operator in their key, so switching it rebuilds the object. */
export function makeGroup(op: GroupOp, items: FilterNode[]): FilterGroup {
if (op === 'or') return { any_of: items }
if (op === 'none') return { none_of: items }
return { all_of: items }
}
frontend/src/lib/components/triggers/kafka/KafkaTriggerEditorInner.svelte
+3 -8
View File
@@ -29,6 +29,7 @@
import TriggerSuspendedJobsAlert from '../TriggerSuspendedJobsAlert.svelte'
import TriggerSuspendedJobsModal from '../TriggerSuspendedJobsModal.svelte'
import TriggerFilters from '../TriggerFilters.svelte'
import type { FilterNode } from '../filters'
import Select from '$lib/components/select/Select.svelte'
import Toggle from '$lib/components/Toggle.svelte'
@@ -102,7 +103,7 @@
let error_handler_path: string | undefined = $state()
let error_handler_args: Record<string, any> = $state({})
let retry: Retry | undefined = $state()
let filters: { key: string; value: any }[] = $state([])
let filters: FilterNode[] = $state([])
let filterLogic = $state<'and' | 'or'>('and')
let suspendedJobsModal = $state<TriggerSuspendedJobsModal | null>(null)
@@ -303,13 +304,7 @@
deploymentLoading = true
const previousPath = initialPath
const cfg = getSaveCfg()
const isSaved = await saveKafkaTriggerFromCfg(
initialPath,
cfg,
edit,
wsId!,
usedTriggerKinds
)
const isSaved = await saveKafkaTriggerFromCfg(initialPath, cfg, edit, wsId!, usedTriggerKinds)
if (isSaved) {
draftSync.discard(previousPath, getSaveCfg())
onUpdate?.(cfg.path)
frontend/src/lib/components/triggers/websocket/WebsocketTriggerEditorInner.svelte
+5 -7
View File
@@ -29,6 +29,7 @@
import type { Schema } from '$lib/common'
import JsonEditor from '$lib/components/JsonEditor.svelte'
import TriggerFilters from '../TriggerFilters.svelte'
import type { FilterNode } from '../filters'
import Toggle from '$lib/components/Toggle.svelte'
import WebsocketEditorConfigSection from './WebsocketEditorConfigSection.svelte'
import { untrack, type Snippet } from 'svelte'
@@ -97,10 +98,7 @@
let pathError = $state('')
let url = $state('')
let dirtyUrl = $state(false)
let filters: {
key: string
value: any
}[] = $state([])
let filters: FilterNode[] = $state([])
let filterLogic = $state<'and' | 'or'>('and')
let initial_messages: WebsocketTriggerInitialMessage[] = $state([])
let url_runnable_args: Record<string, any> | undefined = $state({})
@@ -257,7 +255,7 @@
is_flow = cfg?.is_flow
path = cfg?.path
url = cfg?.url
filters = cfg?.filters
filters = cfg?.filters ?? []
filterLogic = cfg?.filter_logic ?? 'and'
initial_messages = cfg?.initial_messages ?? []
url_runnable_args = cfg?.url_runnable_args
@@ -325,8 +323,8 @@
return {
noDeployed: !!(s as any)?.no_deployed,
overlay: draftFromBackend
? ({ ...deployedTrigger, ...draftFromBackend } as Record<string, any>)
: undefined
? ({ ...deployedTrigger, ...draftFromBackend } as Record<string, any>)
: undefined
}
}
system_prompts/auto-generated/schemas/kafka_trigger.schema.yaml
+51 -7
View File
@@ -44,18 +44,62 @@ properties:
filters:
type: array
items:
type: object
properties:
key:
type: string
value: {}
oneOf:
- type: object
properties:
key:
type: string
value: {}
required:
- key
- value
- type: object
properties:
path:
type: string
description: Dotted path into nested objects, e.g. `a.b.c`. Does not traverse
arrays.
value: {}
required:
- path
- value
- type: object
properties:
any_of:
type: array
items:
type: object
required:
- any_of
- type: object
properties:
all_of:
type: array
items:
type: object
required:
- all_of
- type: object
properties:
none_of:
type: array
items:
type: object
required:
- none_of
description: 'Filters to match incoming messages (only matching messages trigger
the script). Each entry is either a leaf `{key, value}` (top-level field) or
`{path, value}` (dotted path into nested objects), or a group `{any_of: [...]}`
/ `{all_of: [...]}` / `{none_of: [...]}` nesting more entries. Entries at the
top level are combined with `filter_logic`.'
filter_logic:
type: string
enum:
- and
- or
description: Logic to apply when evaluating filters. 'and' requires all filters
to match, 'or' requires any filter to match.
description: Logic to apply when evaluating the top-level filters. 'and' requires
all of them to match, 'or' requires any of them to match. Nested `any_of`/`all_of`/`none_of`
groups carry their own logic.
auto_offset_reset:
type: string
enum:
system_prompts/auto-generated/schemas/mqtt_trigger.schema.yaml
+12
View File
@@ -37,6 +37,18 @@ properties:
type: array
items:
type: object
properties:
qos:
type: string
enum:
- qos0
- qos1
- qos2
topic:
type: string
required:
- qos
- topic
description: Array of MQTT topics to subscribe to, each with topic name and QoS
level
v3_config:
system_prompts/auto-generated/schemas/websocket_trigger.schema.yaml
+77 -10
View File
@@ -37,24 +37,91 @@ properties:
filters:
type: array
items:
type: object
properties:
key:
type: string
value: {}
description: Array of key-value filters to match incoming messages (only matching
messages trigger the script)
oneOf:
- type: object
properties:
key:
type: string
value: {}
required:
- key
- value
- type: object
properties:
path:
type: string
description: Dotted path into nested objects, e.g. `a.b.c`. Does not traverse
arrays.
value: {}
required:
- path
- value
- type: object
properties:
any_of:
type: array
items:
type: object
required:
- any_of
- type: object
properties:
all_of:
type: array
items:
type: object
required:
- all_of
- type: object
properties:
none_of:
type: array
items:
type: object
required:
- none_of
description: 'Filters to match incoming messages (only matching messages trigger
the script). Each entry is either a leaf `{key, value}` (top-level field) or
`{path, value}` (dotted path into nested objects), or a group `{any_of: [...]}`
/ `{all_of: [...]}` / `{none_of: [...]}` nesting more entries. Entries at the
top level are combined with `filter_logic`.'
filter_logic:
type: string
enum:
- and
- or
description: Logic to apply when evaluating filters. 'and' requires all filters
to match, 'or' requires any filter to match.
description: Logic to apply when evaluating the top-level filters. 'and' requires
all of them to match, 'or' requires any of them to match. Nested `any_of`/`all_of`/`none_of`
groups carry their own logic.
initial_messages:
type: array
items:
type: object
oneOf:
- type: object
properties:
raw_message:
type: string
required:
- raw_message
- type: object
properties:
runnable_result:
type: object
properties:
path:
type: string
args:
type: object
description: The arguments to pass to the script or flow
additionalProperties: true
is_flow:
type: boolean
required:
- path
- args
- is_flow
required:
- runnable_result
description: Messages to send immediately after connecting (can be raw strings
or computed by runnables)
url_runnable_args:
system_prompts/generate.py
+6 -6
View File
@@ -904,7 +904,9 @@ def _resolve_schema_refs(schema: dict, backend_schemas: dict, openflow_schemas:
ref = schema['$ref']
ref_name = ref.split('/')[-1]
if ref_name in seen:
return {'type': 'object'}
# Zod cannot express the recursion inline; stay permissive so the nested
# payload survives parsing instead of being stripped as unknown keys.
return {'type': 'object', 'additionalProperties': True}
source = openflow_schemas if 'openflow.openapi.yaml' in ref or ref_name not in backend_schemas else backend_schemas
ref_schema = source.get(ref_name)
@@ -944,15 +946,13 @@ def _apply_zod_metadata(expr: str, schema: dict) -> str:
def _json_schema_to_zod(schema: dict, indent: int = 0) -> str:
schema = schema or {}
if 'oneOf' in schema:
raise ValueError('Unsupported oneOf in workspace tool Zod schema generation')
if 'allOf' in schema:
raise ValueError('Unsupported allOf in workspace tool Zod schema generation')
if 'anyOf' in schema:
variants = schema.get('anyOf') or schema.get('oneOf')
if variants:
expr = "z.union([{}])".format(
', '.join(_json_schema_to_zod(item, indent) for item in schema['anyOf'])
', '.join(_json_schema_to_zod(item, indent) for item in variants)
)
return _apply_zod_metadata(expr, schema)
system_prompts/utils.py
+55 -6
View File
@@ -356,6 +356,29 @@ def extract_options(text: str, option_pattern: re.Pattern) -> list[dict]:
# =============================================================================
def _resolve_refs(schema, all_schemas: dict, openflow_schemas: dict, seen: tuple[str, ...] = ()):
"""Inline named schemas so a documented shape never dangles on a `$ref` the reader
cannot resolve. A recursive schema stops at a bare object on its second visit."""
if isinstance(schema, list):
return [_resolve_refs(item, all_schemas, openflow_schemas, seen) for item in schema]
if not isinstance(schema, dict):
return schema
ref = schema.get('$ref')
if ref:
ref_name = ref.split('/')[-1]
target = all_schemas.get(ref_name) or openflow_schemas.get(ref_name)
if ref_name in seen or not target:
return {'type': 'object'}
return _resolve_refs(target, all_schemas, openflow_schemas, (*seen, ref_name))
return {
key: _resolve_refs(value, all_schemas, openflow_schemas, seen)
for key, value in schema.items()
}
def extract_cli_schema(schema: dict, all_schemas: dict, openflow_schemas: dict | None = None) -> dict:
"""
Transform an OpenAPI schema to CLI format by removing server-managed fields.
@@ -406,6 +429,13 @@ def extract_cli_schema(schema: dict, all_schemas: dict, openflow_schemas: dict |
else:
# Other external reference
result['properties'][key] = {'type': 'object', 'description': value.get('description', f'See {ref_path}')}
elif value.get('type') == 'array' and '$ref' in (value.get('items') or {}):
# An array of a named schema: inline the item shape, otherwise the
# documented type degrades to a bare object
result['properties'][key] = {
**value,
'items': _resolve_refs(value['items'], all_schemas, openflow_schemas),
}
else:
result['properties'][key] = value
@@ -425,6 +455,30 @@ def extract_cli_schema(schema: dict, all_schemas: dict, openflow_schemas: dict |
return result
def _describe_array_items(items: dict) -> dict:
"""Describe array items one level deep. Named schemas the caller could not inline, and
the ones a recursive schema refers back to, collapse to a bare object."""
variants = items.get('oneOf') or items.get('anyOf')
if variants:
return {'oneOf': [_describe_array_items(variant) for variant in variants]}
if '$ref' in items:
return {'type': 'object'}
if items.get('type', 'object') == 'object' and items.get('properties'):
properties = {
key: (
{'type': 'array', 'items': _describe_array_items(value.get('items') or {})}
if value.get('type') == 'array'
else value
)
for key, value in items['properties'].items()
}
described = {'type': 'object', 'properties': properties}
if items.get('required'):
described['required'] = items['required']
return described
return {'type': items.get('type', 'object')}
def format_schema_as_json(schema: dict) -> dict:
"""Convert a CLI schema to a clean JSON Schema representation."""
if not schema or not schema.get('properties'):
@@ -444,13 +498,8 @@ def format_schema_as_json(schema: dict) -> dict:
# Get type
prop_type = value.get('type', 'string')
if prop_type == 'array':
items = value.get('items', {})
prop_def['type'] = 'array'
item_type = items.get('type', 'object')
if item_type == 'object' and items.get('properties'):
prop_def['items'] = {'type': 'object', 'properties': items.get('properties', {})}
else:
prop_def['items'] = {'type': item_type}
prop_def['items'] = _describe_array_items(value.get('items') or {})
elif '$ref' in value:
# For refs, just indicate the type
ref_name = value['$ref'].split('/')[-1]