connector_client/workflow/
validate.rs

1use super::*;
2use serde_json::Value;
3use std::collections::HashSet;
4
5pub const MAX_SPEC_BYTES: usize = 256 * 1024;
6pub const MAX_BINDING_BYTES: usize = 64 * 1024;
7pub const MAX_STEPS: usize = 100;
8pub const MAX_DEADLINE_MS: u64 = 300_000;
9pub const MAX_CONDITION_DEPTH: usize = 8;
10pub const MAX_CONDITION_NODES: usize = 100;
11pub const MAX_LOCATOR_DEPTH: usize = 4;
12
13fn invalid(message: &str) -> WorkflowError {
14    WorkflowError::new("invalid_spec", "validating", message)
15}
16fn nonempty(value: &str, label: &str) -> Result<(), WorkflowError> {
17    if value.trim().is_empty() || value.len() > 1024 {
18        Err(invalid(&format!(
19            "{label} must be non-empty and at most 1024 bytes"
20        )))
21    } else {
22        Ok(())
23    }
24}
25impl WorkflowSpec {
26    pub fn parse(value: &Value) -> Result<Self, WorkflowError> {
27        if serde_json::to_vec(value).map_or(true, |bytes| bytes.len() > MAX_SPEC_BYTES) {
28            return Err(invalid("Workflow spec exceeds 256 KiB"));
29        }
30        // Unsupported modes/condition kinds are capability failures, not fields
31        // accepted and silently ignored. Inspect only schema-owned locations.
32        if value
33            .get("schemaVersion")
34            .and_then(Value::as_u64)
35            .is_some_and(|version| version != 1)
36        {
37            return Err(WorkflowError::new(
38                "unsupported_feature",
39                "validating",
40                "Only schemaVersion 1 is supported",
41            ));
42        }
43        for (field, supported) in [("mode", "strict"), ("schedule", "sequential")] {
44            if value
45                .get(field)
46                .and_then(Value::as_str)
47                .is_some_and(|v| v != supported)
48            {
49                return Err(WorkflowError::new(
50                    "unsupported_feature",
51                    "validating",
52                    format!("Only {supported} {field} is supported"),
53                ));
54            }
55        }
56        if let Some(steps) = value.get("steps").and_then(Value::as_array) {
57            if steps.len() > MAX_STEPS {
58                return Err(invalid("Workflow cannot exceed 100 steps"));
59            }
60            for step in steps {
61                if let Some(target) = step.get("target") {
62                    check_locator_shape(target, 1)?;
63                }
64                if let Some(op) = step.get("op").and_then(Value::as_str) {
65                    if !["click", "fill", "type", "press", "wait", "query", "tool"].contains(&op) {
66                        return Err(WorkflowError::new(
67                            "unsupported_feature",
68                            "validating",
69                            "Unsupported workflow operation",
70                        ));
71                    }
72                }
73                for field in ["expect", "condition"] {
74                    if let Some(condition) = step.get(field) {
75                        check_condition_shape(condition, 1)?;
76                    }
77                }
78            }
79        }
80        if let Some(goal) = value.get("goal") {
81            check_condition_shape(goal, 1)?;
82        }
83        let spec: Self = serde_json::from_value(value.clone())
84            .map_err(|_| invalid("Workflow fields do not match the strict v1 schema"))?;
85        spec.validate()?;
86        Ok(spec)
87    }
88    pub fn validate(&self) -> Result<(), WorkflowError> {
89        if self.schema_version != 1 || self.mode != "strict" || self.schedule != "sequential" {
90            return Err(WorkflowError::new(
91                "unsupported_feature",
92                "validating",
93                "Only strict sequential workflow schemaVersion 1 is supported",
94            ));
95        }
96        nonempty(&self.run_key, "runKey")?;
97        nonempty(&self.window_id, "windowId")?;
98        if self.steps.is_empty() || self.steps.len() > MAX_STEPS {
99            return Err(invalid("Workflow needs between 1 and 100 steps"));
100        }
101        budget(self.deadline_ms, MAX_DEADLINE_MS, "deadlineMs")?;
102        budget(
103            self.defaults.step_timeout_ms,
104            MAX_DEADLINE_MS,
105            "stepTimeoutMs",
106        )?;
107        budget(
108            self.defaults.locator_timeout_ms,
109            MAX_DEADLINE_MS,
110            "locatorTimeoutMs",
111        )?;
112        budget(self.defaults.poll_interval_ms, 1_000, "pollIntervalMs")?;
113        if self.defaults.poll_interval_ms < 10 {
114            return Err(invalid("pollIntervalMs must be at least 10"));
115        }
116        if self.defaults.failure_evidence_grace_ms > 2_000 {
117            return Err(invalid("Failure evidence grace cannot exceed 2000 ms"));
118        }
119        evidence(&self.evidence)?;
120        for (key, value) in &self.inputs {
121            nonempty(key, "input key")?;
122            if serde_json::to_vec(value).map_or(true, |bytes| bytes.len() > MAX_BINDING_BYTES) {
123                return Err(invalid("Input exceeds binding size limit"));
124            }
125        }
126        let mut seen = HashSet::new();
127        for step in &self.steps {
128            nonempty(&step.id, "step id")?;
129            if seen.contains(step.id.as_str()) {
130                return Err(invalid("Step IDs must be unique"));
131            }
132            if let Some(window) = &step.window_id {
133                nonempty(window, "step windowId")?;
134            }
135            if let Some(timeout) = step.timeout_ms {
136                budget(timeout, MAX_DEADLINE_MS, "timeoutMs")?;
137            }
138            if let Some(policy) = &step.evidence {
139                evidence(policy)?;
140            }
141            match &step.op {
142                StepOp::Click { target } | StepOp::Query { target, .. } => {
143                    locator(target, self, &seen, 1)?
144                }
145                StepOp::Fill { target, value } | StepOp::Type { target, value } => {
146                    locator(target, self, &seen, 1)?;
147                    expr(value, self, &seen, true, false)?;
148                }
149                StepOp::Press { target, key } => {
150                    if let Some(target) = target {
151                        locator(target, self, &seen, 1)?;
152                    }
153                    expr(key, self, &seen, true, true)?;
154                }
155                StepOp::Wait {
156                    condition: wait_condition,
157                } => condition(wait_condition, self, &seen, None, 1, &mut 0)?,
158                StepOp::Tool {
159                    tool,
160                    args,
161                    bindings,
162                } => {
163                    nonempty(tool, "tool")?;
164                    if !args.is_object() {
165                        return Err(invalid("tool args must be an object"));
166                    }
167                    for (pointer, value) in bindings {
168                        validate_pointer(pointer)?;
169                        if pointer.is_empty() || args.pointer(pointer).is_none() {
170                            return Err(invalid(
171                                "Tool bindings require existing, non-root argument pointers",
172                            ));
173                        }
174                        expr(value, self, &seen, false, false)?;
175                    }
176                    let keys: Vec<_> = bindings.keys().collect();
177                    for (i, key) in keys.iter().enumerate() {
178                        if keys
179                            .iter()
180                            .skip(i + 1)
181                            .any(|other| other.starts_with(&format!("{key}/")))
182                        {
183                            return Err(invalid("Tool argument binding paths cannot overlap"));
184                        }
185                    }
186                }
187            }
188            if let StepOp::Query {
189                query: QueryKind::Attribute { name },
190                ..
191            } = &step.op
192            {
193                attribute(name)?;
194            }
195            if let Some(expect) = &step.expect {
196                condition(expect, self, &seen, Some(&step.id), 1, &mut 0)?;
197            }
198            seen.insert(step.id.as_str());
199        }
200        if let Some(goal) = &self.goal {
201            condition(goal, self, &seen, None, 1, &mut 0)?;
202        }
203        Ok(())
204    }
205}
206fn check_condition_shape(value: &Value, depth: usize) -> Result<(), WorkflowError> {
207    if depth > MAX_CONDITION_DEPTH {
208        return Err(invalid("Condition depth exceeds 8"));
209    }
210    if let Some(kind) = value.get("kind").and_then(Value::as_str) {
211        if ![
212            "element",
213            "valueEquals",
214            "textContains",
215            "attributeEquals",
216            "result",
217            "all",
218            "any",
219        ]
220        .contains(&kind)
221        {
222            return Err(WorkflowError::new(
223                "unsupported_condition",
224                "validating",
225                "Unsupported workflow condition",
226            ));
227        }
228    }
229    for field in ["transition", "stabilityMs", "correlation", "invocation"] {
230        if value.get(field).is_some() {
231            return Err(WorkflowError::new(
232                "unsupported_condition",
233                "validating",
234                format!("Condition {field} is not supported"),
235            ));
236        }
237    }
238    if let Some(children) = value.get("conditions").and_then(Value::as_array) {
239        for child in children {
240            check_condition_shape(child, depth + 1)?;
241        }
242    }
243    Ok(())
244}
245fn check_locator_shape(value: &Value, depth: usize) -> Result<(), WorkflowError> {
246    if depth > MAX_LOCATOR_DEPTH {
247        return Err(invalid("Locator scope depth exceeds 4"));
248    }
249    if let Some(scope) = value.get("scope") {
250        check_locator_shape(scope, depth + 1)?;
251    }
252    Ok(())
253}
254fn budget(value: u64, max: u64, name: &str) -> Result<(), WorkflowError> {
255    if value == 0 || value > max {
256        Err(invalid(&format!("{name} must be between 1 and {max}")))
257    } else {
258        Ok(())
259    }
260}
261fn evidence(policy: &EvidencePolicy) -> Result<(), WorkflowError> {
262    if policy.success != "summary" || policy.failure != "scoped" {
263        return Err(WorkflowError::new(
264            "unsupported_feature",
265            "validating",
266            "Only summary success and scoped failure evidence are supported",
267        ));
268    }
269    if !(1024..=65536).contains(&policy.max_inline_bytes) {
270        return Err(invalid("maxInlineBytes must be between 1024 and 65536"));
271    }
272    Ok(())
273}
274fn expr(
275    value: &ValueExpr,
276    spec: &WorkflowSpec,
277    seen: &HashSet<&str>,
278    string: bool,
279    non_empty: bool,
280) -> Result<(), WorkflowError> {
281    let literal = match value {
282        ValueExpr::Literal(value) => Some(value),
283        ValueExpr::FromInput(reference) => {
284            nonempty(&reference.key, "input key")?;
285            Some(spec.inputs.get(&reference.key).ok_or_else(|| {
286                WorkflowError::new(
287                    "binding_missing",
288                    "validating",
289                    "Referenced input does not exist",
290                )
291            })?)
292        }
293        ValueExpr::FromStep(reference) => {
294            validate_pointer(&reference.pointer)?;
295            if !seen.contains(reference.step_id.as_str()) {
296                return Err(WorkflowError::new(
297                    "binding_missing",
298                    "validating",
299                    "Step binding must reference a previous step",
300                ));
301            }
302            None
303        }
304    };
305    if let Some(value) = literal {
306        if serde_json::to_vec(value).map_or(true, |bytes| bytes.len() > MAX_BINDING_BYTES) {
307            return Err(invalid("Literal exceeds binding size limit"));
308        }
309        if string && !value.is_string() {
310            return Err(WorkflowError::new(
311                "binding_type_mismatch",
312                "validating",
313                "Expression requires a string value",
314            ));
315        }
316        if non_empty && value.as_str().is_some_and(|v| v.trim().is_empty()) {
317            return Err(WorkflowError::new(
318                "binding_type_mismatch",
319                "validating",
320                "Identity expression requires a non-empty string",
321            ));
322        }
323    }
324    Ok(())
325}
326fn attribute(name: &str) -> Result<(), WorkflowError> {
327    if name.is_empty()
328        || name.len() > 256
329        || !name
330            .chars()
331            .all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | ':' | '.'))
332    {
333        return Err(invalid("Invalid attribute name"));
334    }
335    Ok(())
336}
337fn locator(
338    target: &Locator,
339    spec: &WorkflowSpec,
340    seen: &HashSet<&str>,
341    depth: usize,
342) -> Result<(), WorkflowError> {
343    if depth > MAX_LOCATOR_DEPTH {
344        return Err(invalid("Locator scope depth exceeds 4"));
345    }
346    expr(&target.value, spec, seen, true, true)?;
347    if let Some(name) = &target.name {
348        expr(name, spec, seen, true, false)?;
349    }
350    if let Some(scope) = &target.scope {
351        locator(scope, spec, seen, depth + 1)?;
352    }
353    if let Some(entity) = &target.entity {
354        attribute(&entity.attribute)?;
355        expr(&entity.value, spec, seen, true, true)?;
356    }
357    Ok(())
358}
359fn condition(
360    value: &Condition,
361    spec: &WorkflowSpec,
362    seen: &HashSet<&str>,
363    current: Option<&str>,
364    depth: usize,
365    nodes: &mut usize,
366) -> Result<(), WorkflowError> {
367    *nodes += 1;
368    if depth > MAX_CONDITION_DEPTH || *nodes > MAX_CONDITION_NODES {
369        return Err(invalid("Condition exceeds depth or node limit"));
370    }
371    match value {
372        Condition::Element { target, .. } => locator(target, spec, seen, 1)?,
373        Condition::ValueEquals { target, expected }
374        | Condition::TextContains { target, expected }
375        | Condition::AttributeEquals {
376            target, expected, ..
377        } => {
378            locator(target, spec, seen, 1)?;
379            expr(expected, spec, seen, true, false)?;
380        }
381        Condition::Result {
382            step_id,
383            pointer,
384            operator,
385            expected,
386        } => {
387            validate_pointer(pointer)?;
388            if !seen.contains(step_id.as_str()) && current != Some(step_id.as_str()) {
389                return Err(WorkflowError::new(
390                    "binding_missing",
391                    "validating",
392                    "Result condition must reference an available step",
393                ));
394            }
395            if (*operator == ResultOperator::Eq) != expected.is_some() {
396                return Err(invalid(
397                    "Only eq result conditions require an expected expression",
398                ));
399            }
400            if let Some(expected) = expected {
401                expr(expected, spec, seen, false, false)?;
402            }
403        }
404        Condition::All { conditions } | Condition::Any { conditions } => {
405            if conditions.is_empty() {
406                return Err(invalid("all/any conditions must be non-empty"));
407            }
408            for child in conditions {
409                condition(child, spec, seen, current, depth + 1, nodes)?;
410            }
411        }
412    }
413    if let Condition::AttributeEquals { name, .. } = value {
414        attribute(name)?;
415    }
416    Ok(())
417}
connector_client/workflow/validate.rs

connector_client/workflow/
validate.rs
