1use super::types::{BatchMode, BatchStep, BrowserResult, VerificationPredicate};
7use super::{
8 INTENT_RESOLUTION_SCHEMA_VERSION, IntentConfidence, IntentConstraints, IntentPolicyDecision,
9 IntentScope, SemanticIntentAction, SemanticIntentExecutionRequest, SemanticIntentRequest,
10 SemanticIntentResult, SemanticResolution, SemanticResolutionPolicy, SemanticRouteIdentity,
11 target_fingerprint_digest,
12};
13use serde::de::Error as DeError;
14use serde::{Deserialize, Deserializer, Serialize, Serializer};
15use serde_json::Value;
16use sha2::{Digest, Sha256};
17use std::collections::{BTreeMap, BTreeSet};
18use std::fmt;
19use std::time::{Duration, Instant};
20use url::Url;
21
22pub const WORKFLOW_SCHEMA_VERSION: u32 = 1;
24const MAX_NAME_BYTES: usize = 128;
25const MAX_DESCRIPTION_BYTES: usize = 4 * 1024;
26const MAX_INPUTS: usize = 64;
27const MAX_STEPS: usize = 64;
28const MAX_DURATION_MS: u64 = 15 * 60 * 1_000;
29const MAX_RETRIES: u32 = 8;
30const MAX_EXTRACTED_BYTES: usize = 4 * 1024 * 1024;
31const MAX_TEXT_BYTES: usize = 64 * 1024;
32const MAX_TARGET_BYTES: usize = 1_024;
33const MAX_WAIT_CONDITION_BYTES: usize = 4 * 1024;
34const MAX_STEP_REPETITIONS: u32 = 8;
35const MAX_WORKFLOW_TRACE_EVENTS: usize = 2_048;
36const WORKFLOW_TRACE_SCHEMA_VERSION: u8 = 1;
37const WORKFLOW_CHECKPOINT_SCHEMA_VERSION: u8 = 1;
38const MAX_WORKFLOW_CHECKPOINT_BYTES: usize = 8 * 1024;
39const MAX_CHECKPOINT_HISTORY_STATES: usize = 64;
40const MAX_CHECKPOINT_EXECUTION_IDS: usize = 8;
41const MAX_INTENT_PURPOSE_BYTES: usize = 256;
42
43#[derive(Debug, Clone, Serialize, Deserialize)]
45#[serde(rename_all = "camelCase")]
46pub struct WorkflowDefinition {
47 pub schema_version: u32,
49 pub name: String,
51 pub workflow_version: String,
53 #[serde(default, skip_serializing_if = "Option::is_none")]
54 pub description: Option<String>,
55 pub inputs: BTreeMap<String, WorkflowInput>,
56 pub budgets: WorkflowBudgets,
57 #[serde(default)]
58 pub preconditions: Vec<VerificationPredicate>,
59 pub steps: Vec<WorkflowStep>,
60 pub terminal_condition: VerificationPredicate,
61 pub outputs: BTreeMap<String, WorkflowOutputDeclaration>,
62}
63
64impl WorkflowDefinition {
65 pub fn from_json(input: &str) -> Result<Self, WorkflowValidationError> {
67 let value: Value = serde_json::from_str(input)
68 .map_err(|error| WorkflowValidationError::new("$", format!("invalid JSON: {error}")))?;
69 Self::from_value(value)
70 }
71
72 pub fn from_value(value: Value) -> Result<Self, WorkflowValidationError> {
74 require_object_fields(
75 &value,
76 &[
77 "schemaVersion",
78 "name",
79 "workflowVersion",
80 "inputs",
81 "budgets",
82 "steps",
83 "terminalCondition",
84 "outputs",
85 ],
86 )?;
87 reject_unknown_fields(
88 &value,
89 "$",
90 &[
91 "schemaVersion",
92 "name",
93 "workflowVersion",
94 "description",
95 "inputs",
96 "budgets",
97 "preconditions",
98 "steps",
99 "terminalCondition",
100 "outputs",
101 ],
102 )?;
103 if let Some(inputs) = value.get("inputs").and_then(Value::as_object) {
104 for (name, input) in inputs {
105 reject_unknown_fields(
106 input,
107 &format!("inputs.{name}"),
108 &["valueType", "type", "required", "maxLength", "sensitive"],
109 )?;
110 }
111 }
112 if let Some(budgets) = value.get("budgets") {
113 reject_unknown_fields(
114 budgets,
115 "budgets",
116 &[
117 "maxSteps",
118 "maxDurationMs",
119 "maxRetries",
120 "maxExtractedBytes",
121 ],
122 )?;
123 }
124 if let Some(outputs) = value.get("outputs").and_then(Value::as_object) {
125 for (name, output) in outputs {
126 reject_unknown_fields(
127 output,
128 &format!("outputs.{name}"),
129 &["valueType", "type", "source", "required", "sensitive"],
130 )?;
131 }
132 }
133 if let Some(preconditions) = value.get("preconditions").and_then(Value::as_array) {
134 for (index, predicate) in preconditions.iter().enumerate() {
135 reject_predicate_fields(predicate, &format!("preconditions[{index}]"))?;
136 }
137 }
138 if let Some(predicate) = value.get("terminalCondition") {
139 reject_predicate_fields(predicate, "terminalCondition")?;
140 }
141 if let Some(steps) = value.get("steps").and_then(Value::as_array) {
142 for (index, step) in steps.iter().enumerate() {
143 reject_unknown_fields(
144 step,
145 &format!("steps[{index}]"),
146 &[
147 "id",
148 "action",
149 "intent",
150 "when",
151 "expect",
152 "beforeRetry",
153 "transaction",
154 "idempotencyKey",
155 "maxRetries",
156 "repeat",
157 "url",
158 "timeoutMs",
159 "target",
160 "text",
161 "value",
162 "condition",
163 "dx",
164 "dy",
165 "includeDom",
166 "includeScreenshot",
167 "includeFormValues",
168 ],
169 )?;
170 if let Some(intent) = step.get("intent") {
171 if step.get("action").is_some() {
172 return Err(WorkflowValidationError::new(
173 format!("steps[{index}].action"),
174 "semantic intent steps cannot also declare a batch action",
175 ));
176 }
177 reject_unknown_fields(
178 intent,
179 &format!("steps[{index}].intent"),
180 &[
181 "action",
182 "purpose",
183 "intent",
184 "scope",
185 "constraints",
186 "resolutionPolicy",
187 "value",
188 ],
189 )?;
190 }
191 for field in ["when", "expect", "beforeRetry"] {
192 if let Some(predicate) = step.get(field) {
193 reject_predicate_fields(predicate, &format!("steps[{index}].{field}"))?;
194 }
195 }
196 }
197 }
198 let definition: Self = serde_json::from_value(value).map_err(|error| {
199 WorkflowValidationError::new("$", format!("invalid workflow shape: {error}"))
200 })?;
201 definition.validate()?;
202 Ok(definition)
203 }
204
205 pub fn validate(&self) -> Result<(), WorkflowValidationError> {
207 if self.schema_version != WORKFLOW_SCHEMA_VERSION {
208 return Err(WorkflowValidationError::new(
209 "schemaVersion",
210 format!(
211 "unsupported schema version {}; expected {}",
212 self.schema_version, WORKFLOW_SCHEMA_VERSION
213 ),
214 ));
215 }
216 validate_name("name", &self.name)?;
217 validate_name("workflowVersion", &self.workflow_version)?;
218 if let Some(description) = &self.description {
219 validate_bytes("description", description, 1, MAX_DESCRIPTION_BYTES)?;
220 }
221 if self.inputs.len() > MAX_INPUTS {
222 return Err(WorkflowValidationError::new(
223 "inputs",
224 format!("must contain at most {MAX_INPUTS} entries"),
225 ));
226 }
227 for (name, input) in &self.inputs {
228 validate_map_key("inputs", name)?;
229 input.validate(&format!("inputs.{name}"))?;
230 }
231 self.budgets.validate("budgets")?;
232 if self.steps.is_empty() {
233 return Err(WorkflowValidationError::new(
234 "steps",
235 "must contain at least one step",
236 ));
237 }
238 if self.steps.len() > self.budgets.max_steps as usize {
239 return Err(WorkflowValidationError::new(
240 "steps",
241 "step count exceeds budgets.maxSteps",
242 ));
243 }
244 let expanded_steps: usize = self.steps.iter().map(|step| step.repeat as usize).sum();
245 if expanded_steps > self.budgets.max_steps as usize {
246 return Err(WorkflowValidationError::new(
247 "steps",
248 "expanded repetition count exceeds budgets.maxSteps",
249 ));
250 }
251
252 let mut ids = BTreeSet::new();
253 let mut idempotency_keys = BTreeSet::new();
254 for (index, step) in self.steps.iter().enumerate() {
255 let path = format!("steps[{index}]");
256 step.validate(&path, self.budgets.max_retries)?;
257 if !ids.insert(step.id.as_str()) {
258 return Err(WorkflowValidationError::new(
259 format!("{path}.id"),
260 format!("duplicate step ID {:?}", step.id),
261 ));
262 }
263 if let Some(key) = &step.idempotency_key
264 && !idempotency_keys.insert(key.as_str())
265 {
266 return Err(WorkflowValidationError::new(
267 format!("{path}.idempotencyKey"),
268 format!("duplicate idempotency key {:?}", key),
269 ));
270 }
271 }
272 for (index, predicate) in self.preconditions.iter().enumerate() {
273 validate_predicate(predicate, &format!("preconditions[{index}]"))?;
274 }
275 validate_predicate(&self.terminal_condition, "terminalCondition")?;
276 for (name, output) in &self.outputs {
277 validate_map_key("outputs", name)?;
278 output.validate(&format!("outputs.{name}"))?;
279 }
280 Ok(())
281 }
282
283 pub fn to_canonical_json(&self) -> Result<String, WorkflowValidationError> {
285 self.validate()?;
286 serde_json::to_string(self).map_err(|error| {
287 WorkflowValidationError::new("$", format!("cannot serialize workflow: {error}"))
288 })
289 }
290
291 pub fn validate_inputs(
293 &self,
294 values: &BTreeMap<String, Value>,
295 ) -> Result<(), WorkflowValidationError> {
296 for name in values.keys() {
297 if !self.inputs.contains_key(name) {
298 return Err(WorkflowValidationError::new(
299 format!("inputs.{name}"),
300 "value has no declared input",
301 ));
302 }
303 }
304 for (name, declaration) in &self.inputs {
305 match values.get(name) {
306 Some(value) => declaration.validate_value(&format!("inputs.{name}"), value)?,
307 None if declaration.required => {
308 return Err(WorkflowValidationError::new(
309 format!("inputs.{name}"),
310 "required input is missing",
311 ));
312 }
313 None => {}
314 }
315 }
316 Ok(())
317 }
318
319 pub fn resolve_actions(
323 &self,
324 values: &BTreeMap<String, Value>,
325 ) -> Result<Vec<WorkflowStep>, WorkflowValidationError> {
326 self.validate_inputs(values)?;
327 self.steps
328 .iter()
329 .enumerate()
330 .map(|(index, step)| {
331 let mut resolved = step.clone();
332 if let Some(intent) = &step.intent {
333 resolved.intent = Some(resolve_workflow_intent(
334 intent,
335 values,
336 &format!("steps[{index}].intent"),
337 )?);
338 } else {
339 resolved.action = resolve_batch_step(
340 &step.action,
341 values,
342 &format!("steps[{index}].action"),
343 )?;
344 }
345 resolved.validate(&format!("steps[{index}]"), self.budgets.max_retries)?;
346 Ok(resolved)
347 })
348 .collect()
349 }
350}
351
352#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
354#[serde(rename_all = "camelCase")]
355pub struct WorkflowInput {
356 #[serde(alias = "type")]
357 pub value_type: WorkflowValueType,
358 #[serde(default = "default_true")]
359 pub required: bool,
360 #[serde(default, skip_serializing_if = "Option::is_none")]
361 pub max_length: Option<usize>,
362 #[serde(default, skip_serializing_if = "Option::is_none")]
363 pub sensitive: Option<bool>,
364}
365
366impl WorkflowInput {
367 fn validate(&self, path: &str) -> Result<(), WorkflowValidationError> {
368 if self.max_length == Some(0) || self.max_length.is_some_and(|value| value > MAX_TEXT_BYTES)
369 {
370 return Err(WorkflowValidationError::new(
371 format!("{path}.maxLength"),
372 format!("must be 1..={MAX_TEXT_BYTES}"),
373 ));
374 }
375 Ok(())
376 }
377
378 fn validate_value(&self, path: &str, value: &Value) -> Result<(), WorkflowValidationError> {
379 let valid = match self.value_type {
380 WorkflowValueType::String => value.is_string(),
381 WorkflowValueType::Integer => value.as_i64().is_some() || value.as_u64().is_some(),
382 WorkflowValueType::Number => value.is_number(),
383 WorkflowValueType::Boolean => value.is_boolean(),
384 WorkflowValueType::Url => value
385 .as_str()
386 .is_some_and(|value| Url::parse(value).is_ok()),
387 };
388 if !valid {
389 return Err(WorkflowValidationError::new(
390 path,
391 format!("expected {}", self.value_type),
392 ));
393 }
394 if let Some(max_length) = self.max_length {
395 let length = value
396 .as_str()
397 .map_or_else(|| value.to_string().len(), str::len);
398 if length > max_length {
399 return Err(WorkflowValidationError::new(
400 path,
401 format!("value exceeds maxLength {max_length}"),
402 ));
403 }
404 }
405 Ok(())
406 }
407}
408
409fn default_true() -> bool {
410 true
411}
412
413#[derive(Debug, Clone, Serialize, Deserialize)]
415#[serde(rename_all = "camelCase")]
416pub struct WorkflowBudgets {
417 pub max_steps: u32,
418 pub max_duration_ms: u64,
419 #[serde(default)]
420 pub max_retries: u32,
421 pub max_extracted_bytes: usize,
422}
423
424impl WorkflowBudgets {
425 fn validate(&self, path: &str) -> Result<(), WorkflowValidationError> {
426 if self.max_steps == 0 || self.max_steps as usize > MAX_STEPS {
427 return Err(WorkflowValidationError::new(
428 format!("{path}.maxSteps"),
429 format!("must be 1..={MAX_STEPS}"),
430 ));
431 }
432 if self.max_duration_ms == 0 || self.max_duration_ms > MAX_DURATION_MS {
433 return Err(WorkflowValidationError::new(
434 format!("{path}.maxDurationMs"),
435 format!("must be 1..={MAX_DURATION_MS}"),
436 ));
437 }
438 if self.max_retries > MAX_RETRIES {
439 return Err(WorkflowValidationError::new(
440 format!("{path}.maxRetries"),
441 format!("must be <= {MAX_RETRIES}"),
442 ));
443 }
444 if self.max_extracted_bytes == 0 || self.max_extracted_bytes > MAX_EXTRACTED_BYTES {
445 return Err(WorkflowValidationError::new(
446 format!("{path}.maxExtractedBytes"),
447 format!("must be 1..={MAX_EXTRACTED_BYTES}"),
448 ));
449 }
450 Ok(())
451 }
452}
453
454#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
456#[serde(rename_all = "snake_case")]
457pub enum WorkflowValueType {
458 String,
459 Integer,
460 Number,
461 Boolean,
462 Url,
463}
464
465impl fmt::Display for WorkflowValueType {
466 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
467 formatter.write_str(match self {
468 Self::String => "string",
469 Self::Integer => "integer",
470 Self::Number => "number",
471 Self::Boolean => "boolean",
472 Self::Url => "url",
473 })
474 }
475}
476
477#[derive(Debug, Clone)]
479pub struct WorkflowStep {
480 pub id: String,
481 pub action: BatchStep,
482 pub intent: Option<WorkflowIntentStep>,
483 pub when: Option<VerificationPredicate>,
484 pub expect: Option<VerificationPredicate>,
485 pub before_retry: Option<VerificationPredicate>,
486 pub transaction: WorkflowTransactionClass,
487 pub idempotency_key: Option<String>,
488 pub max_retries: u32,
489 pub repeat: u32,
490}
491
492#[derive(Debug, Clone, Serialize, Deserialize)]
495#[serde(rename_all = "camelCase", deny_unknown_fields)]
496pub struct WorkflowIntentStep {
497 pub action: SemanticIntentAction,
498 #[serde(default, skip_serializing_if = "Option::is_none")]
499 pub purpose: Option<String>,
500 #[serde(default, skip_serializing_if = "Option::is_none")]
501 pub intent: Option<String>,
502 #[serde(default)]
503 pub scope: IntentScope,
504 #[serde(default)]
505 pub constraints: IntentConstraints,
506 #[serde(default = "default_workflow_resolution_policy")]
507 pub resolution_policy: SemanticResolutionPolicy,
508 #[serde(default, skip_serializing_if = "Option::is_none")]
509 pub value: Option<String>,
510}
511
512impl WorkflowIntentStep {
513 fn validate(&self, path: &str) -> Result<(), WorkflowValidationError> {
514 let supplied = self.purpose.is_some() as u8 + self.intent.is_some() as u8;
515 if supplied != 1 {
516 return Err(WorkflowValidationError::new(
517 format!("{path}.purpose"),
518 "provide exactly one of purpose or intent",
519 ));
520 }
521 if let Some(purpose) = &self.purpose {
522 validate_bytes(
523 &format!("{path}.purpose"),
524 purpose,
525 1,
526 MAX_INTENT_PURPOSE_BYTES,
527 )?;
528 }
529 if let Some(intent) = &self.intent {
530 validate_bytes(
531 &format!("{path}.intent"),
532 intent,
533 1,
534 MAX_INTENT_PURPOSE_BYTES * 2,
535 )?;
536 }
537 if let Some(value) = &self.value {
538 validate_bytes(&format!("{path}.value"), value, 0, 4_096)?;
539 }
540 self.execution_request(path).map(|_| ())
541 }
542
543 fn execution_request(
544 &self,
545 path: &str,
546 ) -> Result<SemanticIntentExecutionRequest, WorkflowValidationError> {
547 let intent = self
548 .intent
549 .clone()
550 .or_else(|| self.purpose.as_deref().map(purpose_to_intent))
551 .ok_or_else(|| {
552 WorkflowValidationError::new(format!("{path}.purpose"), "intent phrase is required")
553 })?;
554 let request = SemanticIntentRequest {
555 schema_version: INTENT_RESOLUTION_SCHEMA_VERSION,
556 intent,
557 action: self.action,
558 scope: self.scope.clone(),
559 constraints: self.constraints.clone(),
560 resolution_policy: self.resolution_policy,
561 expected_revision: None,
562 };
563 let execution = SemanticIntentExecutionRequest {
564 request,
565 candidate_id: "workflow-selected-candidate".into(),
566 value: self.value.clone(),
567 };
568 execution.validate().map_err(|error| {
569 WorkflowValidationError::new(format!("{path}.{}", error.path), error.reason)
570 })?;
571 Ok(execution)
572 }
573}
574
575fn default_workflow_resolution_policy() -> SemanticResolutionPolicy {
576 SemanticResolutionPolicy::RequireUniqueHighConfidence
577}
578
579fn purpose_to_intent(purpose: &str) -> String {
580 let mut result = String::with_capacity(purpose.len() + 8);
581 for (index, character) in purpose.chars().enumerate() {
582 if character.is_ascii_uppercase() && index > 0 {
583 result.push(' ');
584 }
585 result.push(character.to_ascii_lowercase());
586 }
587 result
588}
589
590#[derive(Debug, Clone, Serialize, Deserialize)]
592#[serde(rename_all = "camelCase")]
593pub struct WorkflowRecordedTarget {
594 pub role: String,
595 pub name: String,
596 #[serde(default, skip_serializing_if = "Option::is_none")]
597 pub context: Option<String>,
598 #[serde(default, skip_serializing_if = "Option::is_none")]
599 pub region_kind: Option<super::SemanticRegionKind>,
600}
601
602#[derive(Debug, Clone, Serialize, Deserialize)]
608#[serde(rename_all = "camelCase")]
609pub struct WorkflowRecordedRoute {
610 pub target_digest: String,
611 pub frame_digest: String,
612 pub url: String,
613}
614
615#[derive(Debug, Clone, Serialize, Deserialize)]
617#[serde(rename_all = "camelCase")]
618pub struct WorkflowRecordedSemantic {
619 pub intent: String,
620 pub normalized_intent: String,
621 pub action: SemanticIntentAction,
622 pub resolution: SemanticResolution,
623 pub policy_decision: IntentPolicyDecision,
624 pub candidate_count: usize,
625 pub excluded_count: usize,
626 pub ambiguous: bool,
627 #[serde(default, skip_serializing_if = "Option::is_none")]
628 pub revision: Option<u64>,
629 #[serde(default, skip_serializing_if = "Option::is_none")]
630 pub route: Option<WorkflowRecordedRoute>,
631 #[serde(default, skip_serializing_if = "Option::is_none")]
632 pub target_fingerprint: Option<String>,
633}
634
635#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
637#[serde(rename_all = "snake_case")]
638pub enum WorkflowRecordingConfidence {
639 High,
640 Medium,
641 Low,
642}
643
644#[derive(Debug, Clone, Serialize, Deserialize)]
646#[serde(rename_all = "camelCase")]
647pub struct WorkflowDraftStep {
648 pub id: String,
649 pub action: BatchStep,
650 #[serde(skip_serializing_if = "Option::is_none")]
651 pub intent: Option<WorkflowIntentStep>,
652 #[serde(skip_serializing_if = "Option::is_none")]
653 pub target: Option<WorkflowRecordedTarget>,
654 #[serde(skip_serializing_if = "Option::is_none")]
655 pub semantic: Option<WorkflowRecordedSemantic>,
656 #[serde(skip_serializing_if = "Option::is_none")]
657 pub expect: Option<VerificationPredicate>,
658 pub transaction: WorkflowTransactionClass,
659 pub confidence: WorkflowRecordingConfidence,
660 pub review_required: bool,
661 #[serde(skip_serializing_if = "Option::is_none")]
662 pub input_name: Option<String>,
663 #[serde(default)]
664 pub sensitive_input: bool,
665}
666
667#[derive(Debug, Clone, Serialize, Deserialize)]
669#[serde(rename_all = "camelCase")]
670pub struct WorkflowDraft {
671 pub schema_version: u32,
672 pub name: String,
673 pub workflow_version: String,
674 pub steps: Vec<WorkflowDraftStep>,
675}
676
677#[derive(Debug, Clone)]
679pub struct WorkflowRecorder {
680 draft: WorkflowDraft,
681}
682
683impl WorkflowRecorder {
684 pub fn new(name: impl Into<String>, workflow_version: impl Into<String>) -> Self {
687 Self {
688 draft: WorkflowDraft {
689 schema_version: WORKFLOW_SCHEMA_VERSION,
690 name: name.into(),
691 workflow_version: workflow_version.into(),
692 steps: Vec::new(),
693 },
694 }
695 }
696
697 pub fn record_click(
699 &mut self,
700 id: impl Into<String>,
701 role: impl Into<String>,
702 name: impl Into<String>,
703 expect: Option<VerificationPredicate>,
704 ) -> Result<(), WorkflowValidationError> {
705 let target = recorded_target(role.into(), name.into(), None, None)?;
706 let locator = format!("role={};name={}", target.role, target.name);
707 self.push(WorkflowDraftStep {
708 id: id.into(),
709 action: BatchStep::Click { target: locator },
710 intent: None,
711 target: Some(target),
712 semantic: None,
713 expect,
714 transaction: WorkflowTransactionClass::Unknown,
715 confidence: WorkflowRecordingConfidence::High,
716 review_required: true,
717 input_name: None,
718 sensitive_input: false,
719 })
720 }
721
722 pub fn record_type_input(
724 &mut self,
725 id: impl Into<String>,
726 role: impl Into<String>,
727 name: impl Into<String>,
728 input_name: impl Into<String>,
729 ) -> Result<(), WorkflowValidationError> {
730 let target = recorded_target(role.into(), name.into(), None, None)?;
731 let input_name = input_name.into();
732 validate_name("inputName", &input_name)?;
733 let sensitive_input = looks_sensitive_input_name(&input_name);
734 let locator = format!("role={};name={}", target.role, target.name);
735 self.push(WorkflowDraftStep {
736 id: id.into(),
737 action: BatchStep::Type {
738 text: format!("${{inputs.{input_name}}}"),
739 target: Some(locator),
740 },
741 intent: None,
742 target: Some(target),
743 semantic: None,
744 expect: None,
745 transaction: WorkflowTransactionClass::Unknown,
746 confidence: WorkflowRecordingConfidence::High,
747 review_required: true,
748 input_name: Some(input_name),
749 sensitive_input,
750 })
751 }
752
753 pub fn record_observe(&mut self, id: impl Into<String>) -> Result<(), WorkflowValidationError> {
755 self.push(WorkflowDraftStep {
756 id: id.into(),
757 action: BatchStep::Observe {
758 include_dom: false,
759 include_screenshot: false,
760 include_form_values: false,
761 },
762 intent: None,
763 target: None,
764 semantic: None,
765 expect: None,
766 transaction: WorkflowTransactionClass::ReadOnly,
767 confidence: WorkflowRecordingConfidence::High,
768 review_required: true,
769 input_name: None,
770 sensitive_input: false,
771 })
772 }
773
774 pub fn record_semantic_intent(
780 &mut self,
781 id: impl Into<String>,
782 request: &SemanticIntentRequest,
783 result: &SemanticIntentResult,
784 input_name: Option<impl Into<String>>,
785 transaction: WorkflowTransactionClass,
786 expect: Option<VerificationPredicate>,
787 ) -> Result<(), WorkflowValidationError> {
788 request
789 .validate()
790 .map_err(|error| WorkflowValidationError::new("semantic.request", error.to_string()))?;
791 result
792 .validate()
793 .map_err(|error| WorkflowValidationError::new("semantic.result", error.to_string()))?;
794 if request.action != result.action || request.intent != result.intent {
795 return Err(WorkflowValidationError::new(
796 "semantic",
797 "request and result action/intent do not match",
798 ));
799 }
800
801 let input_name = input_name.map(Into::into);
802 let value = match request.action {
803 SemanticIntentAction::Type | SemanticIntentAction::Select => {
804 let input_name = input_name.as_deref().ok_or_else(|| {
805 WorkflowValidationError::new(
806 "inputName",
807 "type and select recordings require an input name",
808 )
809 })?;
810 validate_name("inputName", input_name)?;
811 Some(format!("${{inputs.{input_name}}}"))
812 }
813 _ if input_name.is_some() => {
814 return Err(WorkflowValidationError::new(
815 "inputName",
816 "only type and select recordings accept an input name",
817 ));
818 }
819 _ => None,
820 };
821
822 let selected = result.selected_candidate.as_deref().and_then(|id| {
823 result
824 .candidates
825 .iter()
826 .find(|candidate| candidate.id == id)
827 });
828 let target = selected
829 .map(|candidate| {
830 recorded_target(
831 candidate.role.clone(),
832 candidate.name.clone(),
833 candidate.region_kind.map(|kind| format!("{kind:?}")),
834 candidate.region_kind,
835 )
836 })
837 .transpose()?;
838 let target_fingerprint = selected.and_then(|candidate| {
839 candidate.fingerprint.as_ref().map(|fingerprint| {
840 target_fingerprint_digest(
841 &candidate.role,
842 &candidate.name,
843 candidate.input_type.as_deref(),
844 candidate.region_kind,
845 fingerprint.purpose,
846 )
847 })
848 });
849 let confidence = selected
850 .map(|candidate| recording_confidence(candidate.confidence))
851 .unwrap_or(WorkflowRecordingConfidence::Low);
852 let semantic = WorkflowRecordedSemantic {
853 intent: result.intent.clone(),
854 normalized_intent: result.normalized_intent.clone(),
855 action: result.action,
856 resolution: result.resolution,
857 policy_decision: result.policy_decision,
858 candidate_count: result.candidates.len(),
859 excluded_count: result.excluded_count,
860 ambiguous: matches!(result.resolution, SemanticResolution::Ambiguous),
861 revision: result.revision,
862 route: result.route.as_ref().map(recorded_route),
863 target_fingerprint,
864 };
865 let intent = WorkflowIntentStep {
866 action: request.action,
867 purpose: None,
868 intent: Some(request.intent.clone()),
869 scope: request.scope.clone(),
870 constraints: request.constraints.clone(),
871 resolution_policy: request.resolution_policy,
872 value,
873 };
874 self.push(WorkflowDraftStep {
875 id: id.into(),
876 action: BatchStep::Observe {
877 include_dom: false,
878 include_screenshot: false,
879 include_form_values: false,
880 },
881 intent: Some(intent),
882 target,
883 semantic: Some(semantic),
884 expect,
885 transaction,
886 confidence,
887 review_required: true,
888 sensitive_input: input_name
889 .as_deref()
890 .is_some_and(looks_sensitive_input_name),
891 input_name,
892 })
893 }
894
895 pub fn draft(&self) -> &WorkflowDraft {
896 &self.draft
897 }
898
899 pub fn inferred_inputs(&self) -> BTreeMap<String, WorkflowInput> {
905 let mut inputs = BTreeMap::new();
906 for step in &self.draft.steps {
907 let Some(name) = &step.input_name else {
908 continue;
909 };
910 let sensitive = step.sensitive_input;
911 inputs
912 .entry(name.clone())
913 .and_modify(|input: &mut WorkflowInput| {
914 if sensitive {
915 input.sensitive = Some(true);
916 }
917 })
918 .or_insert_with(|| WorkflowInput {
919 value_type: WorkflowValueType::String,
920 required: true,
921 max_length: None,
922 sensitive: sensitive.then_some(true),
923 });
924 }
925 inputs
926 }
927
928 pub fn into_definition(
930 self,
931 inputs: BTreeMap<String, WorkflowInput>,
932 budgets: WorkflowBudgets,
933 terminal_condition: VerificationPredicate,
934 outputs: BTreeMap<String, WorkflowOutputDeclaration>,
935 ) -> Result<WorkflowDefinition, WorkflowValidationError> {
936 let definition = WorkflowDefinition {
937 schema_version: self.draft.schema_version,
938 name: self.draft.name,
939 workflow_version: self.draft.workflow_version,
940 description: Some("Recorded draft; review before execution.".into()),
941 inputs,
942 budgets,
943 preconditions: Vec::new(),
944 steps: self
945 .draft
946 .steps
947 .into_iter()
948 .map(|step| WorkflowStep {
949 id: step.id,
950 action: step.action,
951 intent: step.intent,
952 when: None,
953 expect: step.expect,
954 before_retry: None,
955 transaction: step.transaction,
956 idempotency_key: None,
957 max_retries: 0,
958 repeat: 1,
959 })
960 .collect(),
961 terminal_condition,
962 outputs,
963 };
964 definition.validate()?;
965 Ok(definition)
966 }
967
968 fn push(&mut self, step: WorkflowDraftStep) -> Result<(), WorkflowValidationError> {
969 if self.draft.steps.len() >= MAX_STEPS {
970 return Err(WorkflowValidationError::new(
971 "steps",
972 format!("must contain at most {MAX_STEPS} entries"),
973 ));
974 }
975 validate_name("steps.id", &step.id)?;
976 if self.draft.steps.iter().any(|item| item.id == step.id) {
977 return Err(WorkflowValidationError::new(
978 "steps.id",
979 format!("duplicate step ID {:?}", step.id),
980 ));
981 }
982 self.draft.steps.push(step);
983 Ok(())
984 }
985}
986
987fn recorded_target(
988 role: String,
989 name: String,
990 context: Option<String>,
991 region_kind: Option<super::SemanticRegionKind>,
992) -> Result<WorkflowRecordedTarget, WorkflowValidationError> {
993 validate_bytes("target.role", &role, 1, 128)?;
994 validate_bytes("target.name", &name, 1, MAX_TARGET_BYTES)?;
995 if role.contains([';', '\n', '\r']) || name.contains([';', '\n', '\r']) {
996 return Err(WorkflowValidationError::new(
997 "target",
998 "semantic target fields cannot contain locator separators or newlines",
999 ));
1000 }
1001 if let Some(context) = &context {
1002 validate_bytes("target.context", context, 1, 256)?;
1003 }
1004 Ok(WorkflowRecordedTarget {
1005 role,
1006 name,
1007 context,
1008 region_kind,
1009 })
1010}
1011
1012fn recording_confidence(confidence: IntentConfidence) -> WorkflowRecordingConfidence {
1013 match confidence {
1014 IntentConfidence::Exact | IntentConfidence::High => WorkflowRecordingConfidence::High,
1015 IntentConfidence::Medium => WorkflowRecordingConfidence::Medium,
1016 IntentConfidence::Low | IntentConfidence::Insufficient => WorkflowRecordingConfidence::Low,
1017 }
1018}
1019
1020fn looks_sensitive_input_name(name: &str) -> bool {
1021 let normalized = name.to_ascii_lowercase();
1022 [
1023 "password", "passwd", "secret", "token", "api_key", "apikey", "cookie",
1024 ]
1025 .iter()
1026 .any(|term| normalized.contains(term))
1027}
1028
1029fn recorded_route(route: &SemanticRouteIdentity) -> WorkflowRecordedRoute {
1030 let url = Url::parse(&route.url)
1031 .map(|mut parsed| {
1032 let _ = parsed.set_username("");
1033 let _ = parsed.set_password(None);
1034 parsed.set_query(None);
1035 parsed.set_fragment(None);
1036 parsed.to_string()
1037 })
1038 .unwrap_or_else(|_| bound_workflow_text(&route.url, 2_048));
1039 WorkflowRecordedRoute {
1040 target_digest: hash_recorded_identifier(&route.target_id),
1041 frame_digest: hash_recorded_identifier(&route.frame_id),
1042 url,
1043 }
1044}
1045
1046fn hash_recorded_identifier(value: &str) -> String {
1047 let digest = Sha256::digest(value.as_bytes());
1048 format!("sha256:{digest:x}")
1049}
1050
1051#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
1054#[serde(rename_all = "snake_case")]
1055pub enum WorkflowTransactionClass {
1056 ReadOnly,
1057 Idempotent,
1058 ConditionallyIdempotent,
1059 NonIdempotent,
1060 #[default]
1061 Unknown,
1062}
1063
1064impl WorkflowTransactionClass {
1065 pub fn permits_pre_dispatch_retry(self) -> bool {
1068 matches!(
1069 self,
1070 Self::ReadOnly | Self::Idempotent | Self::ConditionallyIdempotent
1071 )
1072 }
1073}
1074
1075impl Serialize for WorkflowStep {
1076 fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
1077 where
1078 S: Serializer,
1079 {
1080 let mut action = if let Some(intent) = &self.intent {
1081 serde_json::json!({ "intent": intent })
1082 } else {
1083 serde_json::to_value(&self.action).map_err(serde::ser::Error::custom)?
1084 };
1085 if self.intent.is_none() {
1086 let object = action
1087 .as_object_mut()
1088 .ok_or_else(|| serde::ser::Error::custom("workflow action must be an object"))?;
1089 for (internal, public) in [
1090 ("timeout_ms", "timeoutMs"),
1091 ("include_dom", "includeDom"),
1092 ("include_screenshot", "includeScreenshot"),
1093 ("include_form_values", "includeFormValues"),
1094 ] {
1095 if let Some(value) = object.remove(internal) {
1096 object.insert(public.to_string(), value);
1097 }
1098 }
1099 }
1100
1101 let mut workflow = serde_json::Map::new();
1102 workflow.insert("id".into(), Value::String(self.id.clone()));
1103 if let Value::Object(action) = action {
1104 workflow.extend(action);
1105 }
1106 if let Some(when) = &self.when {
1107 workflow.insert(
1108 "when".into(),
1109 serde_json::to_value(when).map_err(serde::ser::Error::custom)?,
1110 );
1111 }
1112 if let Some(expect) = &self.expect {
1113 workflow.insert(
1114 "expect".into(),
1115 serde_json::to_value(expect).map_err(serde::ser::Error::custom)?,
1116 );
1117 }
1118 if let Some(before_retry) = &self.before_retry {
1119 workflow.insert(
1120 "beforeRetry".into(),
1121 serde_json::to_value(before_retry).map_err(serde::ser::Error::custom)?,
1122 );
1123 }
1124 workflow.insert(
1125 "transaction".into(),
1126 serde_json::to_value(self.transaction).map_err(serde::ser::Error::custom)?,
1127 );
1128 if let Some(key) = &self.idempotency_key {
1129 workflow.insert("idempotencyKey".into(), Value::String(key.clone()));
1130 }
1131 if self.max_retries > 0 {
1132 workflow.insert("maxRetries".into(), Value::from(self.max_retries));
1133 }
1134 if self.repeat > 1 {
1135 workflow.insert("repeat".into(), Value::from(self.repeat));
1136 }
1137 Value::Object(workflow).serialize(serializer)
1138 }
1139}
1140
1141impl<'de> Deserialize<'de> for WorkflowStep {
1142 fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
1143 where
1144 D: Deserializer<'de>,
1145 {
1146 let mut workflow = serde_json::Map::<String, Value>::deserialize(deserializer)?;
1147 let id = workflow
1148 .remove("id")
1149 .ok_or_else(|| D::Error::custom("workflow step is missing id"))?;
1150 let id = serde_json::from_value(id).map_err(D::Error::custom)?;
1151 let when = workflow
1152 .remove("when")
1153 .map(serde_json::from_value)
1154 .transpose()
1155 .map_err(D::Error::custom)?;
1156 let expect = workflow
1157 .remove("expect")
1158 .map(serde_json::from_value)
1159 .transpose()
1160 .map_err(D::Error::custom)?;
1161 let before_retry = workflow
1162 .remove("beforeRetry")
1163 .map(serde_json::from_value)
1164 .transpose()
1165 .map_err(D::Error::custom)?;
1166 let transaction = workflow
1167 .remove("transaction")
1168 .map(serde_json::from_value)
1169 .transpose()
1170 .map_err(D::Error::custom)?
1171 .unwrap_or_default();
1172 let idempotency_key = workflow
1173 .remove("idempotencyKey")
1174 .map(serde_json::from_value)
1175 .transpose()
1176 .map_err(D::Error::custom)?;
1177 let max_retries = workflow
1178 .remove("maxRetries")
1179 .map(serde_json::from_value)
1180 .transpose()
1181 .map_err(D::Error::custom)?
1182 .unwrap_or(0);
1183 let repeat = workflow
1184 .remove("repeat")
1185 .map(serde_json::from_value)
1186 .transpose()
1187 .map_err(D::Error::custom)?
1188 .unwrap_or(1);
1189 let intent = workflow
1190 .remove("intent")
1191 .map(serde_json::from_value)
1192 .transpose()
1193 .map_err(D::Error::custom)?;
1194 for (public, internal) in [
1195 ("timeoutMs", "timeout_ms"),
1196 ("includeDom", "include_dom"),
1197 ("includeScreenshot", "include_screenshot"),
1198 ("includeFormValues", "include_form_values"),
1199 ] {
1200 if let Some(value) = workflow.remove(public) {
1201 workflow.insert(internal.into(), value);
1202 }
1203 }
1204 let action = if intent.is_some() {
1205 BatchStep::Observe {
1206 include_dom: false,
1207 include_screenshot: false,
1208 include_form_values: false,
1209 }
1210 } else {
1211 serde_json::from_value(Value::Object(workflow)).map_err(D::Error::custom)?
1212 };
1213 Ok(Self {
1214 id,
1215 action,
1216 intent,
1217 when,
1218 expect,
1219 before_retry,
1220 transaction,
1221 idempotency_key,
1222 max_retries,
1223 repeat,
1224 })
1225 }
1226}
1227
1228impl WorkflowStep {
1229 fn validate(
1230 &self,
1231 path: &str,
1232 workflow_max_retries: u32,
1233 ) -> Result<(), WorkflowValidationError> {
1234 validate_name(&format!("{path}.id"), &self.id)?;
1235 if let Some(intent) = &self.intent {
1236 intent.validate(&format!("{path}.intent"))?;
1237 } else {
1238 validate_batch_step(&self.action, &format!("{path}.action"))?;
1239 }
1240 if let Some(predicate) = &self.when {
1241 validate_predicate(predicate, &format!("{path}.when"))?;
1242 }
1243 if let Some(predicate) = &self.expect {
1244 validate_predicate(predicate, &format!("{path}.expect"))?;
1245 }
1246 if let Some(predicate) = &self.before_retry {
1247 validate_predicate(predicate, &format!("{path}.beforeRetry"))?;
1248 }
1249 if let Some(key) = &self.idempotency_key {
1250 validate_bytes(&format!("{path}.idempotencyKey"), key, 1, 256)?;
1251 }
1252 if self.repeat == 0 || self.repeat > MAX_STEP_REPETITIONS {
1253 return Err(WorkflowValidationError::new(
1254 format!("{path}.repeat"),
1255 format!("must be 1..={MAX_STEP_REPETITIONS}"),
1256 ));
1257 }
1258 if self.repeat > 1 && self.when.is_some() {
1259 return Err(WorkflowValidationError::new(
1260 format!("{path}.repeat"),
1261 "conditional steps cannot repeat automatically",
1262 ));
1263 }
1264 if self.repeat > 1 && !self.transaction.permits_pre_dispatch_retry() {
1265 return Err(WorkflowValidationError::new(
1266 format!("{path}.repeat"),
1267 "unknown or non-idempotent steps cannot repeat automatically",
1268 ));
1269 }
1270 if self.max_retries > workflow_max_retries {
1271 return Err(WorkflowValidationError::new(
1272 format!("{path}.maxRetries"),
1273 "step retry count exceeds budgets.maxRetries",
1274 ));
1275 }
1276 match self.transaction {
1277 WorkflowTransactionClass::ConditionallyIdempotent if self.idempotency_key.is_none() => {
1278 return Err(WorkflowValidationError::new(
1279 format!("{path}.idempotencyKey"),
1280 "conditionally idempotent steps require an idempotency key",
1281 ));
1282 }
1283 WorkflowTransactionClass::NonIdempotent | WorkflowTransactionClass::Unknown
1284 if self.max_retries > 0 =>
1285 {
1286 return Err(WorkflowValidationError::new(
1287 format!("{path}.maxRetries"),
1288 "non-idempotent or unknown steps cannot be retried automatically",
1289 ));
1290 }
1291 _ => {}
1292 }
1293 Ok(())
1294 }
1295}
1296
1297#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
1299#[serde(rename_all = "snake_case")]
1300pub enum WorkflowStepState {
1301 Pending,
1302 Ready,
1303 Preflight,
1304 Resolving,
1305 NotDispatched,
1306 Dispatched,
1307 EffectObserved,
1308 Verified,
1309 OutputsExtracted,
1310 Committed,
1311 FailedBeforeDispatch,
1312 FailedAfterDispatch,
1313 Indeterminate,
1314 Skipped,
1315}
1316
1317impl WorkflowStepState {
1318 pub fn can_transition_to(self, next: Self) -> bool {
1320 matches!(
1321 (self, next),
1322 (Self::Pending, Self::Ready | Self::Skipped)
1323 | (Self::Ready, Self::Preflight | Self::Skipped)
1324 | (
1325 Self::Preflight,
1326 Self::Resolving | Self::EffectObserved | Self::FailedBeforeDispatch
1327 )
1328 | (Self::Resolving, Self::NotDispatched | Self::Dispatched)
1329 | (Self::NotDispatched, Self::FailedBeforeDispatch)
1330 | (
1331 Self::Dispatched,
1332 Self::EffectObserved | Self::FailedAfterDispatch | Self::Indeterminate
1333 )
1334 | (
1335 Self::EffectObserved,
1336 Self::Verified | Self::FailedAfterDispatch | Self::Indeterminate
1337 )
1338 | (Self::Verified, Self::OutputsExtracted)
1339 | (Self::OutputsExtracted, Self::Committed)
1340 | (Self::FailedBeforeDispatch, Self::Ready)
1341 | (Self::Committed, Self::Ready)
1342 )
1343 }
1344}
1345
1346#[derive(Debug, Clone, Serialize, Deserialize)]
1348#[serde(rename_all = "camelCase")]
1349pub struct WorkflowStepRecord {
1350 pub id: String,
1351 pub state: WorkflowStepState,
1352 pub history: Vec<WorkflowStepState>,
1353 pub attempts: u32,
1354 #[serde(default, skip_serializing_if = "Vec::is_empty")]
1355 pub execution_ids: Vec<String>,
1356 #[serde(default)]
1357 pub dispatch_acknowledged: bool,
1358 #[serde(default)]
1359 pub effect_observed: bool,
1360 #[serde(default)]
1361 pub postcondition_verified: bool,
1362 #[serde(default)]
1363 pub retry_safe: bool,
1364 #[serde(default, skip_serializing_if = "Option::is_none")]
1365 pub previous_revision: Option<u64>,
1366 #[serde(default, skip_serializing_if = "Option::is_none")]
1367 pub current_revision: Option<u64>,
1368 #[serde(skip_serializing_if = "Option::is_none")]
1369 pub branch_decision: Option<WorkflowBranchDecision>,
1370 #[serde(skip_serializing_if = "Option::is_none")]
1371 pub intent_evidence: Option<WorkflowIntentEvidence>,
1372 #[serde(skip_serializing_if = "Option::is_none")]
1373 pub error: Option<String>,
1374}
1375
1376#[derive(Debug, Clone, Serialize, Deserialize)]
1378#[serde(rename_all = "camelCase")]
1379pub struct WorkflowIntentEvidence {
1380 pub resolution_id: String,
1381 pub candidate_id: String,
1382 pub revision: u64,
1383 pub resolution: super::SemanticResolution,
1384 pub policy_decision: super::IntentPolicyDecision,
1385 pub confidence: super::IntentConfidence,
1386 pub fingerprint: Option<super::SemanticTargetFingerprint>,
1387}
1388
1389impl WorkflowStepRecord {
1390 fn new(id: &str) -> Self {
1391 Self {
1392 id: id.to_string(),
1393 state: WorkflowStepState::Pending,
1394 history: vec![WorkflowStepState::Pending],
1395 attempts: 0,
1396 execution_ids: Vec::new(),
1397 dispatch_acknowledged: false,
1398 effect_observed: false,
1399 postcondition_verified: false,
1400 retry_safe: false,
1401 previous_revision: None,
1402 current_revision: None,
1403 branch_decision: None,
1404 intent_evidence: None,
1405 error: None,
1406 }
1407 }
1408
1409 fn transition(&mut self, next: WorkflowStepState) -> Result<(), String> {
1410 if !self.state.can_transition_to(next) {
1411 return Err(format!(
1412 "invalid workflow step transition {} -> {}",
1413 state_name(self.state),
1414 state_name(next)
1415 ));
1416 }
1417 self.state = next;
1418 self.history.push(next);
1419 Ok(())
1420 }
1421
1422 fn fail(&mut self, state: WorkflowStepState, error: &str) {
1423 let _ = self.transition(state);
1424 self.error = Some(bound_workflow_text(error, 512));
1425 }
1426}
1427
1428#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
1430#[serde(rename_all = "snake_case")]
1431pub enum WorkflowRunStatus {
1432 Completed,
1433 Failed,
1434 BudgetExhausted,
1435 ResumeRequired,
1436}
1437
1438#[derive(Debug, Clone, Serialize, Deserialize)]
1440#[serde(rename_all = "camelCase")]
1441pub struct WorkflowTerminalProof {
1442 pub predicate: VerificationPredicate,
1443 pub revision: u64,
1444 pub state: String,
1445}
1446
1447#[derive(Debug, Clone, Serialize, Deserialize)]
1449#[serde(rename_all = "camelCase")]
1450pub struct WorkflowRunResult {
1451 pub run_id: String,
1452 pub name: String,
1453 pub workflow_version: String,
1454 pub status: WorkflowRunStatus,
1455 pub steps: Vec<WorkflowStepRecord>,
1456 pub trace: WorkflowTrace,
1457 pub outputs: BTreeMap<String, WorkflowOutput>,
1458 #[serde(skip_serializing_if = "Option::is_none")]
1459 pub terminal_proof: Option<WorkflowTerminalProof>,
1460 #[serde(skip_serializing_if = "Option::is_none")]
1461 pub failed_step: Option<String>,
1462 #[serde(skip_serializing_if = "Option::is_none")]
1463 pub failure: Option<String>,
1464 pub initial_revision: u64,
1465 pub final_revision: u64,
1466}
1467
1468#[derive(Debug, Clone, Serialize, Deserialize)]
1470#[serde(rename_all = "camelCase")]
1471pub struct WorkflowBranchDecision {
1472 pub step_id: String,
1473 pub predicate: VerificationPredicate,
1474 pub matched: bool,
1475}
1476
1477#[derive(Debug, Clone, Serialize, Deserialize)]
1479#[serde(rename_all = "camelCase")]
1480pub struct WorkflowTrace {
1481 #[serde(default = "default_workflow_trace_schema_version")]
1482 pub schema_version: u8,
1483 #[serde(default, skip_serializing_if = "Option::is_none")]
1484 pub run_id: Option<String>,
1485 pub events: Vec<WorkflowTraceEvent>,
1486 #[serde(default, skip_serializing_if = "Vec::is_empty")]
1487 pub branch_decisions: Vec<WorkflowBranchDecision>,
1488 #[serde(default, skip_serializing_if = "Vec::is_empty")]
1489 pub intent_resolutions: Vec<WorkflowIntentEvidence>,
1490}
1491
1492#[derive(Debug, Clone, Serialize, Deserialize)]
1494#[serde(rename_all = "camelCase")]
1495pub struct WorkflowTraceEvent {
1496 pub sequence: u64,
1497 pub step_id: String,
1498 pub state: WorkflowStepState,
1499 pub attempt: u32,
1500}
1501
1502impl WorkflowTrace {
1503 pub fn from_steps(steps: &[WorkflowStepRecord]) -> Self {
1505 let mut events = Vec::new();
1506 for step in steps {
1507 let mut attempt = 0u32;
1508 for state in &step.history {
1509 if events.len() == MAX_WORKFLOW_TRACE_EVENTS {
1510 break;
1511 }
1512 if *state == WorkflowStepState::Preflight {
1513 attempt = attempt.saturating_add(1);
1514 }
1515 events.push(WorkflowTraceEvent {
1516 sequence: events.len() as u64,
1517 step_id: step.id.clone(),
1518 state: *state,
1519 attempt,
1520 });
1521 }
1522 }
1523 let branch_decisions = steps
1524 .iter()
1525 .filter_map(|step| step.branch_decision.clone())
1526 .collect();
1527 let intent_resolutions = steps
1528 .iter()
1529 .filter_map(|step| step.intent_evidence.clone())
1530 .collect();
1531 Self {
1532 schema_version: WORKFLOW_TRACE_SCHEMA_VERSION,
1533 run_id: None,
1534 events,
1535 branch_decisions,
1536 intent_resolutions,
1537 }
1538 }
1539
1540 pub fn validate(&self) -> Result<(), WorkflowValidationError> {
1542 if self.schema_version != WORKFLOW_TRACE_SCHEMA_VERSION {
1543 return Err(WorkflowValidationError::new(
1544 "trace.schemaVersion",
1545 format!(
1546 "unsupported trace schema version {}; expected {}",
1547 self.schema_version, WORKFLOW_TRACE_SCHEMA_VERSION
1548 ),
1549 ));
1550 }
1551 if self
1552 .run_id
1553 .as_ref()
1554 .is_some_and(|run_id| run_id.is_empty() || run_id.len() > 128)
1555 {
1556 return Err(WorkflowValidationError::new(
1557 "trace.runId",
1558 "run ID must contain 1 to 128 bytes",
1559 ));
1560 }
1561 if self.events.len() > MAX_WORKFLOW_TRACE_EVENTS {
1562 return Err(WorkflowValidationError::new(
1563 "trace.events",
1564 format!("must contain at most {MAX_WORKFLOW_TRACE_EVENTS} events"),
1565 ));
1566 }
1567 for (index, event) in self.events.iter().enumerate() {
1568 if event.sequence != index as u64 {
1569 return Err(WorkflowValidationError::new(
1570 format!("trace.events[{index}].sequence"),
1571 "sequence must be contiguous from zero",
1572 ));
1573 }
1574 }
1575 for (index, decision) in self.branch_decisions.iter().enumerate() {
1576 if decision.step_id.is_empty() {
1577 return Err(WorkflowValidationError::new(
1578 format!("trace.branchDecisions[{index}].stepId"),
1579 "step ID must not be empty",
1580 ));
1581 }
1582 validate_predicate(
1583 &decision.predicate,
1584 &format!("trace.branchDecisions[{index}].predicate"),
1585 )?;
1586 }
1587 Ok(())
1588 }
1589
1590 pub fn replay(
1597 &self,
1598 workflow: &WorkflowDefinition,
1599 ) -> Result<Vec<WorkflowStepRecord>, WorkflowValidationError> {
1600 workflow.validate()?;
1601 self.validate()?;
1602 let mut records: Vec<_> = workflow
1603 .steps
1604 .iter()
1605 .map(|step| WorkflowStepRecord::new(&step.id))
1606 .collect();
1607 let mut seen = vec![false; records.len()];
1608 let mut highest_step = 0usize;
1609 let mut started = false;
1610
1611 for (event_index, event) in self.events.iter().enumerate() {
1612 let step_index = workflow
1613 .steps
1614 .iter()
1615 .position(|step| step.id == event.step_id)
1616 .ok_or_else(|| {
1617 WorkflowValidationError::new(
1618 format!("trace.events[{event_index}].stepId"),
1619 "step ID is not declared by the workflow",
1620 )
1621 })?;
1622 if !started {
1623 if step_index != 0 {
1624 return Err(WorkflowValidationError::new(
1625 format!("trace.events[{event_index}].stepId"),
1626 "trace must begin with the first declared step",
1627 ));
1628 }
1629 started = true;
1630 }
1631 if step_index > highest_step.saturating_add(1) {
1632 return Err(WorkflowValidationError::new(
1633 format!("trace.events[{event_index}].stepId"),
1634 "trace skips a declared step",
1635 ));
1636 }
1637 if step_index < highest_step {
1638 return Err(WorkflowValidationError::new(
1639 format!("trace.events[{event_index}].stepId"),
1640 "trace returns to an earlier step",
1641 ));
1642 }
1643 highest_step = highest_step.max(step_index);
1644
1645 let record = &mut records[step_index];
1646 if !seen[step_index] {
1647 if event.state != WorkflowStepState::Pending || event.attempt != 0 {
1648 return Err(WorkflowValidationError::new(
1649 format!("trace.events[{event_index}]"),
1650 "each step must begin with pending at attempt zero",
1651 ));
1652 }
1653 seen[step_index] = true;
1654 continue;
1655 }
1656 if event.state == WorkflowStepState::Preflight {
1657 if event.attempt != record.attempts.saturating_add(1) {
1658 return Err(WorkflowValidationError::new(
1659 format!("trace.events[{event_index}].attempt"),
1660 "preflight attempt must increment by one",
1661 ));
1662 }
1663 record.attempts = event.attempt;
1664 } else if event.attempt != record.attempts {
1665 return Err(WorkflowValidationError::new(
1666 format!("trace.events[{event_index}].attempt"),
1667 "event attempt does not match the current step attempt",
1668 ));
1669 }
1670 record.transition(event.state).map_err(|reason| {
1671 WorkflowValidationError::new(format!("trace.events[{event_index}].state"), reason)
1672 })?;
1673 }
1674 for decision in &self.branch_decisions {
1675 let index = workflow
1676 .steps
1677 .iter()
1678 .position(|step| step.id == decision.step_id)
1679 .ok_or_else(|| {
1680 WorkflowValidationError::new(
1681 "trace.branchDecisions",
1682 "branch decision references an undeclared step",
1683 )
1684 })?;
1685 if workflow.steps[index].when.as_ref() != Some(&decision.predicate) {
1686 return Err(WorkflowValidationError::new(
1687 "trace.branchDecisions",
1688 "branch decision predicate does not match the workflow",
1689 ));
1690 }
1691 records[index].branch_decision = Some(decision.clone());
1692 }
1693 Ok(records)
1694 }
1695}
1696
1697fn default_workflow_trace_schema_version() -> u8 {
1698 WORKFLOW_TRACE_SCHEMA_VERSION
1699}
1700
1701#[derive(Debug, Clone, Serialize, Deserialize)]
1704#[serde(rename_all = "camelCase")]
1705pub struct WorkflowCheckpoint {
1706 pub schema_version: u8,
1707 #[serde(default)]
1708 pub run_id: String,
1709 pub workflow_name: String,
1710 pub workflow_version: String,
1711 pub definition_hash: String,
1712 pub status: WorkflowRunStatus,
1713 pub next_step_index: usize,
1714 pub steps: Vec<WorkflowCheckpointStep>,
1715 pub page: WorkflowCheckpointPage,
1716}
1717
1718#[derive(Debug, Clone, Serialize, Deserialize)]
1720#[serde(rename_all = "camelCase")]
1721pub struct WorkflowCheckpointStep {
1722 pub id: String,
1723 pub state: WorkflowStepState,
1724 pub attempts: u32,
1725 #[serde(default, skip_serializing_if = "Vec::is_empty")]
1726 pub history: Vec<WorkflowStepState>,
1727 #[serde(default, skip_serializing_if = "Vec::is_empty")]
1728 pub execution_ids: Vec<String>,
1729 #[serde(default)]
1730 pub dispatch_acknowledged: bool,
1731 #[serde(default)]
1732 pub effect_observed: bool,
1733 #[serde(default)]
1734 pub postcondition_verified: bool,
1735 #[serde(default)]
1736 pub retry_safe: bool,
1737 #[serde(default, skip_serializing_if = "Option::is_none")]
1738 pub previous_revision: Option<u64>,
1739 #[serde(default, skip_serializing_if = "Option::is_none")]
1740 pub current_revision: Option<u64>,
1741 #[serde(default, skip_serializing_if = "Option::is_none")]
1742 pub branch_decision: Option<WorkflowBranchDecision>,
1743 #[serde(default, skip_serializing_if = "Option::is_none")]
1744 pub intent_evidence: Option<WorkflowIntentEvidence>,
1745}
1746
1747#[derive(Debug, Clone, Serialize, Deserialize)]
1749#[serde(rename_all = "camelCase")]
1750pub struct WorkflowCheckpointPage {
1751 pub target_id: String,
1752 pub frame_id: String,
1753 pub url: String,
1754 pub title: String,
1755 pub revision: u64,
1756}
1757
1758#[derive(Debug, Clone, Serialize, Deserialize)]
1760#[serde(rename_all = "camelCase")]
1761pub struct WorkflowResumePlan {
1762 pub workflow_name: String,
1763 pub workflow_version: String,
1764 pub next_step_index: usize,
1765 pub current_revision: u64,
1766 pub reconciled: bool,
1767}
1768
1769#[derive(Debug, Clone, Serialize)]
1771#[serde(tag = "kind", rename_all = "snake_case")]
1772pub enum WorkflowResumeError {
1773 SchemaVersionMismatch {
1774 expected: u8,
1775 found: u8,
1776 },
1777 DefinitionMismatch,
1778 RouteChanged,
1779 InvalidState {
1780 step_id: String,
1781 state: WorkflowStepState,
1782 },
1783 CheckpointTooLarge,
1784 CheckpointShape(String),
1785}
1786
1787impl fmt::Display for WorkflowResumeError {
1788 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1789 match self {
1790 Self::SchemaVersionMismatch { expected, found } => {
1791 write!(
1792 formatter,
1793 "workflow checkpoint schema mismatch: expected {expected}, found {found}"
1794 )
1795 }
1796 Self::DefinitionMismatch => {
1797 formatter.write_str("workflow definition does not match checkpoint")
1798 }
1799 Self::RouteChanged => {
1800 formatter.write_str("workflow checkpoint route or target changed")
1801 }
1802 Self::InvalidState { step_id, state } => {
1803 write!(
1804 formatter,
1805 "workflow step {step_id:?} cannot be resumed from {state:?}"
1806 )
1807 }
1808 Self::CheckpointTooLarge => {
1809 formatter.write_str("workflow checkpoint exceeds the 8 KiB limit")
1810 }
1811 Self::CheckpointShape(message) => {
1812 write!(formatter, "invalid workflow checkpoint: {message}")
1813 }
1814 }
1815 }
1816}
1817
1818impl std::error::Error for WorkflowResumeError {}
1819
1820impl WorkflowRunResult {
1821 fn failed(
1822 workflow: &WorkflowDefinition,
1823 run_id: String,
1824 steps: Vec<WorkflowStepRecord>,
1825 failed_step: Option<String>,
1826 failure: impl Into<String>,
1827 initial_revision: u64,
1828 final_revision: u64,
1829 ) -> Self {
1830 let mut trace = WorkflowTrace::from_steps(&steps);
1831 trace.run_id = Some(run_id.clone());
1832 Self {
1833 run_id,
1834 name: workflow.name.clone(),
1835 workflow_version: workflow.workflow_version.clone(),
1836 status: WorkflowRunStatus::Failed,
1837 steps,
1838 trace,
1839 outputs: BTreeMap::new(),
1840 terminal_proof: None,
1841 failed_step,
1842 failure: Some(bound_workflow_text(&failure.into(), 512)),
1843 initial_revision,
1844 final_revision,
1845 }
1846 }
1847
1848 fn budget_exhausted(
1849 workflow: &WorkflowDefinition,
1850 run_id: String,
1851 steps: Vec<WorkflowStepRecord>,
1852 failed_step: Option<String>,
1853 reason: impl Into<String>,
1854 initial_revision: u64,
1855 final_revision: u64,
1856 ) -> Self {
1857 let mut trace = WorkflowTrace::from_steps(&steps);
1858 trace.run_id = Some(run_id.clone());
1859 Self {
1860 run_id,
1861 name: workflow.name.clone(),
1862 workflow_version: workflow.workflow_version.clone(),
1863 status: WorkflowRunStatus::BudgetExhausted,
1864 steps,
1865 trace,
1866 outputs: BTreeMap::new(),
1867 terminal_proof: None,
1868 failed_step,
1869 failure: Some(bound_workflow_text(&reason.into(), 512)),
1870 initial_revision,
1871 final_revision,
1872 }
1873 }
1874
1875 fn resume_required(
1876 workflow: &WorkflowDefinition,
1877 run_id: String,
1878 steps: Vec<WorkflowStepRecord>,
1879 failed_step: Option<String>,
1880 reason: impl Into<String>,
1881 initial_revision: u64,
1882 final_revision: u64,
1883 ) -> Self {
1884 let mut trace = WorkflowTrace::from_steps(&steps);
1885 trace.run_id = Some(run_id.clone());
1886 Self {
1887 run_id,
1888 name: workflow.name.clone(),
1889 workflow_version: workflow.workflow_version.clone(),
1890 status: WorkflowRunStatus::ResumeRequired,
1891 steps,
1892 trace,
1893 outputs: BTreeMap::new(),
1894 terminal_proof: None,
1895 failed_step,
1896 failure: Some(bound_workflow_text(&reason.into(), 512)),
1897 initial_revision,
1898 final_revision,
1899 }
1900 }
1901}
1902
1903impl super::BrowserSession {
1904 async fn execute_workflow_intent_step(
1905 &self,
1906 intent: &WorkflowIntentStep,
1907 expected_revision: u64,
1908 ) -> BrowserResult<(super::types::BatchOutcome, WorkflowIntentEvidence)> {
1909 let mut execution = intent.execution_request("workflow.intent")?;
1910 execution.request.expected_revision = Some(expected_revision);
1911 let resolution = self.resolve_intent(&execution.request).await?;
1912 let candidate_id = resolution.selected_candidate.clone().ok_or_else(|| {
1913 format!(
1914 "semantic workflow intent was not uniquely authorized: resolution={}, policy={}",
1915 serde_json::to_string(&resolution.resolution).unwrap_or_else(|_| "unknown".into()),
1916 serde_json::to_string(&resolution.policy_decision)
1917 .unwrap_or_else(|_| "unknown".into())
1918 )
1919 })?;
1920 execution.candidate_id = candidate_id;
1921 let result = self.execute_intent(&execution).await?;
1922 let accepted_candidate = result
1923 .resolution
1924 .candidates
1925 .iter()
1926 .find(|candidate| candidate.id == result.candidate_id)
1927 .ok_or("executed intent result omitted its accepted candidate")?;
1928 let evidence = WorkflowIntentEvidence {
1929 resolution_id: result.resolution_id.clone(),
1930 candidate_id: result.candidate_id.clone(),
1931 revision: result
1932 .resolution
1933 .revision
1934 .ok_or("executed intent result omitted revision")?,
1935 resolution: result.resolution.resolution,
1936 policy_decision: result.resolution.policy_decision,
1937 confidence: accepted_candidate.confidence,
1938 fingerprint: accepted_candidate.fingerprint.clone(),
1939 };
1940 let Some(action) = result.action else {
1941 return Err(result
1942 .reason
1943 .unwrap_or_else(|| "semantic workflow intent was not executed".into())
1944 .into());
1945 };
1946 let execution_id = action.execution_id.clone();
1947 Ok((
1948 super::types::BatchOutcome {
1949 mode: BatchMode::Fixed,
1950 initial_revision: expected_revision,
1951 final_revision: current_revision(self),
1952 steps: vec![super::types::BatchStepOutcome::Success {
1953 index: 0,
1954 action: "intent".into(),
1955 response_bytes: serde_json::to_string(&action).ok().map(|value| value.len()),
1956 execution_id: Some(execution_id),
1957 }],
1958 completed: 1,
1959 failed: 0,
1960 total: 1,
1961 success: true,
1962 },
1963 evidence,
1964 ))
1965 }
1966
1967 pub async fn run_workflow(
1970 &self,
1971 workflow: &WorkflowDefinition,
1972 inputs: &BTreeMap<String, Value>,
1973 ) -> BrowserResult<WorkflowRunResult> {
1974 workflow.validate()?;
1975 workflow.validate_inputs(inputs)?;
1976 let resolved_steps = workflow.resolve_actions(inputs)?;
1977 let initial_revision = self
1978 .page_revision
1979 .load(std::sync::atomic::Ordering::Relaxed);
1980 let run_id = format!("run_{}", self.next_execution_id());
1981 let started = Instant::now();
1982 let duration_budget = Duration::from_millis(workflow.budgets.max_duration_ms);
1983 let mut executed_steps = 0u32;
1984 let mut records: Vec<_> = resolved_steps
1985 .iter()
1986 .map(|step| WorkflowStepRecord::new(&step.id))
1987 .collect();
1988
1989 for predicate in &workflow.preconditions {
1990 if workflow_budget_expired(started, duration_budget) {
1991 skip_remaining(&mut records, 0);
1992 return Ok(WorkflowRunResult::budget_exhausted(
1993 workflow,
1994 run_id.clone(),
1995 records,
1996 None,
1997 "workflow maxDurationMs budget exhausted before a precondition",
1998 initial_revision,
1999 current_revision(self),
2000 ));
2001 }
2002 if let Err(error) = self
2003 .verify(
2004 predicate.clone(),
2005 workflow_budget_remaining(started, duration_budget),
2006 )
2007 .await
2008 {
2009 skip_remaining(&mut records, 0);
2010 if workflow_budget_expired(started, duration_budget) {
2011 return Ok(WorkflowRunResult::budget_exhausted(
2012 workflow,
2013 run_id.clone(),
2014 records,
2015 None,
2016 "workflow maxDurationMs budget exhausted while checking a precondition",
2017 initial_revision,
2018 current_revision(self),
2019 ));
2020 }
2021 return Ok(WorkflowRunResult::failed(
2022 workflow,
2023 run_id.clone(),
2024 records,
2025 None,
2026 format!("workflow precondition failed: {error}"),
2027 initial_revision,
2028 current_revision(self),
2029 ));
2030 }
2031 }
2032
2033 for (index, step) in resolved_steps.iter().enumerate() {
2034 for repetition in 0..step.repeat {
2035 if executed_steps >= workflow.budgets.max_steps
2036 || workflow_budget_expired(started, duration_budget)
2037 {
2038 skip_remaining(&mut records, index);
2039 return Ok(WorkflowRunResult::budget_exhausted(
2040 workflow,
2041 run_id.clone(),
2042 records,
2043 Some(step.id.clone()),
2044 if executed_steps >= workflow.budgets.max_steps {
2045 "workflow maxSteps budget exhausted before dispatch"
2046 } else {
2047 "workflow maxDurationMs budget exhausted before dispatch"
2048 },
2049 initial_revision,
2050 current_revision(self),
2051 ));
2052 }
2053 executed_steps = executed_steps.saturating_add(1);
2054 if repetition > 0 {
2055 let record = &mut records[index];
2056 let _ = record.transition(WorkflowStepState::Ready);
2057 }
2058 if let Some(predicate) = &step.when {
2059 {
2060 let record = &mut records[index];
2061 if record.state == WorkflowStepState::Pending {
2062 let _ = record.transition(WorkflowStepState::Ready);
2063 }
2064 }
2065 match self.evaluate_predicate_once(predicate).await {
2066 Ok((matched, _state)) => {
2067 let record = &mut records[index];
2068 record.branch_decision = Some(WorkflowBranchDecision {
2069 step_id: step.id.clone(),
2070 predicate: predicate.clone(),
2071 matched,
2072 });
2073 if !matched {
2074 let _ = record.transition(WorkflowStepState::Skipped);
2075 break;
2076 }
2077 }
2078 Err(error) => {
2079 let message = bound_workflow_text(&error.to_string(), 512);
2080 let record = &mut records[index];
2081 let _ = record.transition(WorkflowStepState::Preflight);
2082 record.fail(WorkflowStepState::FailedBeforeDispatch, &message);
2083 skip_remaining(&mut records, index + 1);
2084 if workflow_budget_expired(started, duration_budget) {
2085 return Ok(WorkflowRunResult::budget_exhausted(
2086 workflow,
2087 run_id.clone(),
2088 records,
2089 Some(step.id.clone()),
2090 "workflow maxDurationMs budget exhausted while evaluating a branch",
2091 initial_revision,
2092 current_revision(self),
2093 ));
2094 }
2095 return Ok(WorkflowRunResult::failed(
2096 workflow,
2097 run_id.clone(),
2098 records,
2099 Some(step.id.clone()),
2100 message,
2101 initial_revision,
2102 current_revision(self),
2103 ));
2104 }
2105 }
2106 }
2107 let mut attempt_number = 0;
2108 let mut effect_marker_completed = false;
2109 let mut intent_evidence = None;
2110 let outcome = loop {
2111 let attempt_revision = current_revision(self);
2112 {
2113 let record = &mut records[index];
2114 if record.previous_revision.is_none() {
2115 record.previous_revision = Some(attempt_revision);
2116 }
2117 record.retry_safe = step.transaction.permits_pre_dispatch_retry();
2118 if record.state == WorkflowStepState::Pending {
2119 let _ = record.transition(WorkflowStepState::Ready);
2120 }
2121 let _ = record.transition(WorkflowStepState::Preflight);
2122 let _ = record.transition(WorkflowStepState::Resolving);
2123 attempt_number += 1;
2124 record.attempts = record.attempts.saturating_add(1);
2125 }
2126
2127 let outcome = if let Some(intent) = &step.intent {
2128 match self
2129 .execute_workflow_intent_step(intent, attempt_revision)
2130 .await
2131 {
2132 Ok((outcome, evidence)) => {
2133 intent_evidence = Some(evidence);
2134 Ok(outcome)
2135 }
2136 Err(error) => Err(error),
2137 }
2138 } else {
2139 match workflow_target(&step.action) {
2140 Some(target) => match self.resolve_element(target).await {
2141 Ok(_) => {
2142 self.run_batch_with_mode(
2143 std::slice::from_ref(&step.action),
2144 false,
2145 BatchMode::Unguarded,
2146 None,
2147 )
2148 .await
2149 }
2150 Err(error) => Err(error),
2151 },
2152 None => {
2153 self.run_batch_with_mode(
2154 std::slice::from_ref(&step.action),
2155 false,
2156 BatchMode::Unguarded,
2157 None,
2158 )
2159 .await
2160 }
2161 }
2162 };
2163 match outcome {
2164 Ok(outcome) => break outcome,
2165 Err(error) => {
2166 let message = bound_workflow_text(&error.to_string(), 512);
2167 let retry = can_retry_before_dispatch(
2168 step.transaction,
2169 false,
2170 attempt_number,
2171 step.max_retries,
2172 );
2173 {
2174 let record = &mut records[index];
2175 record.current_revision = Some(current_revision(self));
2176 let _ = record.transition(WorkflowStepState::NotDispatched);
2177 record.fail(WorkflowStepState::FailedBeforeDispatch, &message);
2178 if retry {
2179 let _ = record.transition(WorkflowStepState::Ready);
2180 }
2181 }
2182 if retry {
2183 let marker_matches = match &step.before_retry {
2184 Some(predicate) => {
2185 match self.evaluate_predicate_once(predicate).await {
2186 Ok((matched, _)) => Some(matched),
2187 Err(error) => {
2188 let message = bound_workflow_text(
2189 &format!(
2190 "effect marker could not be evaluated: {error}"
2191 ),
2192 512,
2193 );
2194 let record = &mut records[index];
2195 let _ =
2196 record.transition(WorkflowStepState::Preflight);
2197 record.fail(
2198 WorkflowStepState::FailedBeforeDispatch,
2199 &message,
2200 );
2201 skip_remaining(&mut records, index + 1);
2202 return Ok(WorkflowRunResult::failed(
2203 workflow,
2204 run_id.clone(),
2205 records,
2206 Some(step.id.clone()),
2207 message,
2208 initial_revision,
2209 current_revision(self),
2210 ));
2211 }
2212 }
2213 }
2214 None => None,
2215 };
2216 if marker_matches == Some(true) {
2217 effect_marker_completed = true;
2218 break super::types::BatchOutcome {
2219 mode: BatchMode::Unguarded,
2220 initial_revision: current_revision(self),
2221 final_revision: current_revision(self),
2222 steps: Vec::new(),
2223 completed: 0,
2224 failed: 0,
2225 total: 0,
2226 success: true,
2227 };
2228 }
2229 continue;
2230 }
2231 let message = records[index]
2232 .error
2233 .clone()
2234 .unwrap_or_else(|| "workflow step failed".into());
2235 skip_remaining(&mut records, index + 1);
2236 return Ok(WorkflowRunResult::failed(
2237 workflow,
2238 run_id.clone(),
2239 records,
2240 Some(step.id.clone()),
2241 message,
2242 initial_revision,
2243 current_revision(self),
2244 ));
2245 }
2246 }
2247 };
2248
2249 if effect_marker_completed {
2250 commit_workflow_effect_marker(&mut records[index], current_revision(self));
2251 continue;
2252 }
2253
2254 if !outcome.success {
2255 let message = outcome
2256 .steps
2257 .last()
2258 .and_then(|step| match step {
2259 super::types::BatchStepOutcome::Error { message, .. } => {
2260 Some(message.as_str())
2261 }
2262 super::types::BatchStepOutcome::Success { .. } => None,
2263 })
2264 .unwrap_or("workflow action failed after dispatch");
2265 let message = {
2266 let record = &mut records[index];
2267 record.dispatch_acknowledged = true;
2268 record.current_revision = Some(current_revision(self));
2269 let _ = record.transition(WorkflowStepState::Dispatched);
2270 record.fail(WorkflowStepState::Indeterminate, message);
2271 record
2272 .error
2273 .clone()
2274 .unwrap_or_else(|| "workflow step failed".into())
2275 };
2276 skip_remaining(&mut records, index + 1);
2277 return Ok(WorkflowRunResult::resume_required(
2278 workflow,
2279 run_id.clone(),
2280 records,
2281 Some(step.id.clone()),
2282 message,
2283 initial_revision,
2284 current_revision(self),
2285 ));
2286 }
2287
2288 {
2289 records[index].intent_evidence = intent_evidence;
2290 for execution_id in outcome.steps.iter().filter_map(|step| match step {
2291 super::types::BatchStepOutcome::Success {
2292 execution_id: Some(execution_id),
2293 ..
2294 } => Some(execution_id.clone()),
2295 _ => None,
2296 }) {
2297 records[index].execution_ids.push(execution_id);
2298 }
2299 let record = &mut records[index];
2300 record.dispatch_acknowledged = true;
2301 record.effect_observed = true;
2302 record.current_revision = Some(current_revision(self));
2303 let _ = record.transition(WorkflowStepState::Dispatched);
2304 let _ = record.transition(WorkflowStepState::EffectObserved);
2305 }
2306 if let Some(predicate) = &step.expect
2307 && let Err(error) = self
2308 .verify(
2309 predicate.clone(),
2310 workflow_budget_remaining(started, duration_budget),
2311 )
2312 .await
2313 {
2314 let message = {
2315 let record = &mut records[index];
2316 record.current_revision = Some(current_revision(self));
2317 record.fail(WorkflowStepState::FailedAfterDispatch, &error.to_string());
2318 record
2319 .error
2320 .clone()
2321 .unwrap_or_else(|| "workflow verification failed".into())
2322 };
2323 skip_remaining(&mut records, index + 1);
2324 if workflow_budget_expired(started, duration_budget) {
2325 return Ok(WorkflowRunResult::budget_exhausted(
2326 workflow,
2327 run_id.clone(),
2328 records,
2329 Some(step.id.clone()),
2330 "workflow maxDurationMs budget exhausted while verifying a step",
2331 initial_revision,
2332 current_revision(self),
2333 ));
2334 }
2335 return Ok(WorkflowRunResult::resume_required(
2336 workflow,
2337 run_id.clone(),
2338 records,
2339 Some(step.id.clone()),
2340 message,
2341 initial_revision,
2342 current_revision(self),
2343 ));
2344 }
2345 let record = &mut records[index];
2346 record.postcondition_verified = true;
2347 let _ = record.transition(WorkflowStepState::Verified);
2348 let _ = record.transition(WorkflowStepState::OutputsExtracted);
2349 let _ = record.transition(WorkflowStepState::Committed);
2350 }
2351 }
2352
2353 if workflow_budget_expired(started, duration_budget) {
2354 return Ok(WorkflowRunResult::budget_exhausted(
2355 workflow,
2356 run_id.clone(),
2357 records,
2358 None,
2359 "workflow maxDurationMs budget exhausted before terminal verification",
2360 initial_revision,
2361 current_revision(self),
2362 ));
2363 }
2364 let terminal_proof = match self
2365 .verify(
2366 workflow.terminal_condition.clone(),
2367 workflow_budget_remaining(started, duration_budget),
2368 )
2369 .await
2370 {
2371 Ok(outcome) => WorkflowTerminalProof {
2372 predicate: outcome.predicate,
2373 revision: current_revision(self),
2374 state: outcome.state,
2375 },
2376 Err(error) => {
2377 if workflow_budget_expired(started, duration_budget) {
2378 return Ok(WorkflowRunResult::budget_exhausted(
2379 workflow,
2380 run_id.clone(),
2381 records,
2382 None,
2383 "workflow maxDurationMs budget exhausted while verifying the terminal condition",
2384 initial_revision,
2385 current_revision(self),
2386 ));
2387 }
2388 return Ok(WorkflowRunResult::failed(
2389 workflow,
2390 run_id.clone(),
2391 records,
2392 None,
2393 format!("workflow terminal condition was not proven: {error}"),
2394 initial_revision,
2395 current_revision(self),
2396 ));
2397 }
2398 };
2399 if workflow_budget_expired(started, duration_budget) {
2400 return Ok(WorkflowRunResult::budget_exhausted(
2401 workflow,
2402 run_id.clone(),
2403 records,
2404 None,
2405 "workflow maxDurationMs budget exhausted before output extraction",
2406 initial_revision,
2407 current_revision(self),
2408 ));
2409 }
2410 let outputs = match extract_workflow_outputs(self, workflow).await {
2411 Ok(outputs) => outputs,
2412 Err(error) => {
2413 if workflow_budget_expired(started, duration_budget) {
2414 return Ok(WorkflowRunResult::budget_exhausted(
2415 workflow,
2416 run_id.clone(),
2417 records,
2418 None,
2419 "workflow maxDurationMs budget exhausted while extracting outputs",
2420 initial_revision,
2421 current_revision(self),
2422 ));
2423 }
2424 return Ok(WorkflowRunResult::failed(
2425 workflow,
2426 run_id.clone(),
2427 records,
2428 None,
2429 format!("workflow output extraction failed: {error}"),
2430 initial_revision,
2431 current_revision(self),
2432 ));
2433 }
2434 };
2435 if workflow_budget_expired(started, duration_budget) {
2436 return Ok(WorkflowRunResult::budget_exhausted(
2437 workflow,
2438 run_id.clone(),
2439 records,
2440 None,
2441 "workflow maxDurationMs budget exhausted after output extraction",
2442 initial_revision,
2443 current_revision(self),
2444 ));
2445 }
2446
2447 let mut trace = WorkflowTrace::from_steps(&records);
2448 trace.run_id = Some(run_id.clone());
2449 Ok(WorkflowRunResult {
2450 run_id,
2451 name: workflow.name.clone(),
2452 workflow_version: workflow.workflow_version.clone(),
2453 status: WorkflowRunStatus::Completed,
2454 steps: records,
2455 trace,
2456 outputs,
2457 terminal_proof: Some(terminal_proof),
2458 failed_step: None,
2459 failure: None,
2460 initial_revision,
2461 final_revision: current_revision(self),
2462 })
2463 }
2464}
2465
2466impl super::BrowserSession {
2467 pub async fn export_workflow_checkpoint(
2469 &self,
2470 workflow: &WorkflowDefinition,
2471 result: &WorkflowRunResult,
2472 ) -> BrowserResult<WorkflowCheckpoint> {
2473 workflow.validate()?;
2474 if result.name != workflow.name
2475 || result.workflow_version != workflow.workflow_version
2476 || result.steps.len() != workflow.steps.len()
2477 {
2478 return Err(WorkflowResumeError::CheckpointShape(
2479 "run result does not belong to workflow definition".into(),
2480 )
2481 .into());
2482 }
2483 let page = self.page_info().await?;
2484 let next_step_index = result
2485 .steps
2486 .iter()
2487 .position(|step| step.state != WorkflowStepState::Committed)
2488 .unwrap_or(result.steps.len());
2489 let checkpoint = WorkflowCheckpoint {
2490 schema_version: WORKFLOW_CHECKPOINT_SCHEMA_VERSION,
2491 run_id: result.run_id.clone(),
2492 workflow_name: workflow.name.clone(),
2493 workflow_version: workflow.workflow_version.clone(),
2494 definition_hash: workflow_definition_hash(workflow)?,
2495 status: result.status,
2496 next_step_index,
2497 steps: result
2498 .steps
2499 .iter()
2500 .map(|step| WorkflowCheckpointStep {
2501 id: step.id.clone(),
2502 state: step.state,
2503 attempts: step.attempts,
2504 history: step.history.clone(),
2505 execution_ids: step.execution_ids.clone(),
2506 dispatch_acknowledged: step.dispatch_acknowledged,
2507 effect_observed: step.effect_observed,
2508 postcondition_verified: step.postcondition_verified,
2509 retry_safe: step.retry_safe,
2510 previous_revision: step.previous_revision,
2511 current_revision: step.current_revision,
2512 branch_decision: step.branch_decision.clone(),
2513 intent_evidence: step.intent_evidence.clone(),
2514 })
2515 .collect(),
2516 page: WorkflowCheckpointPage {
2517 target_id: bound_workflow_text(&page.target_id, 256),
2518 frame_id: bound_workflow_text(&page.frame_id, 256),
2519 url: bound_workflow_text(&page.url, 1_024),
2520 title: bound_workflow_text(&page.title, 1_024),
2521 revision: current_revision(self),
2522 },
2523 };
2524 checkpoint.validate_size()?;
2525 Ok(checkpoint)
2526 }
2527
2528 pub fn parse_workflow_checkpoint(
2530 input: &str,
2531 ) -> Result<WorkflowCheckpoint, WorkflowResumeError> {
2532 let checkpoint: WorkflowCheckpoint = serde_json::from_str(input)
2533 .map_err(|error| WorkflowResumeError::CheckpointShape(error.to_string()))?;
2534 checkpoint.validate_size()?;
2535 Ok(checkpoint)
2536 }
2537
2538 pub async fn reconcile_workflow_checkpoint(
2541 &self,
2542 workflow: &WorkflowDefinition,
2543 checkpoint: &WorkflowCheckpoint,
2544 ) -> BrowserResult<WorkflowResumePlan> {
2545 workflow.validate()?;
2546 checkpoint.validate_size()?;
2547 if checkpoint.schema_version != WORKFLOW_CHECKPOINT_SCHEMA_VERSION {
2548 return Err(WorkflowResumeError::SchemaVersionMismatch {
2549 expected: WORKFLOW_CHECKPOINT_SCHEMA_VERSION,
2550 found: checkpoint.schema_version,
2551 }
2552 .into());
2553 }
2554 if checkpoint.workflow_name != workflow.name
2555 || checkpoint.workflow_version != workflow.workflow_version
2556 || checkpoint.definition_hash != workflow_definition_hash(workflow)?
2557 {
2558 return Err(WorkflowResumeError::DefinitionMismatch.into());
2559 }
2560 if checkpoint.steps.len() != workflow.steps.len()
2561 || checkpoint
2562 .steps
2563 .iter()
2564 .zip(&workflow.steps)
2565 .any(|(checkpoint_step, step)| checkpoint_step.id != step.id)
2566 {
2567 return Err(WorkflowResumeError::CheckpointShape(
2568 "checkpoint steps do not match workflow steps".into(),
2569 )
2570 .into());
2571 }
2572
2573 let page = self.page_info().await?;
2574 if checkpoint.page.target_id != page.target_id
2575 || checkpoint.page.frame_id != page.frame_id
2576 || checkpoint.page.url != bound_workflow_text(&page.url, 1_024)
2577 || checkpoint.page.title != bound_workflow_text(&page.title, 1_024)
2578 {
2579 return Err(WorkflowResumeError::RouteChanged.into());
2580 }
2581
2582 let mut next_step_index = checkpoint.steps.len();
2583 for (index, checkpoint_step) in checkpoint.steps.iter().enumerate() {
2584 if checkpoint_step.state != WorkflowStepState::Committed {
2585 next_step_index = index;
2586 break;
2587 }
2588 }
2589 if checkpoint.next_step_index != next_step_index {
2590 return Err(WorkflowResumeError::CheckpointShape(
2591 "nextStepIndex does not match step states".into(),
2592 )
2593 .into());
2594 }
2595 if checkpoint.status == WorkflowRunStatus::Completed
2596 && next_step_index != workflow.steps.len()
2597 {
2598 return Err(WorkflowResumeError::CheckpointShape(
2599 "completed checkpoint has uncommitted steps".into(),
2600 )
2601 .into());
2602 }
2603 if next_step_index < workflow.steps.len() {
2604 let step = &workflow.steps[next_step_index];
2605 let state = checkpoint.steps[next_step_index].state;
2606 if state == WorkflowStepState::FailedBeforeDispatch
2607 && !step.transaction.permits_pre_dispatch_retry()
2608 {
2609 return Err(WorkflowResumeError::InvalidState {
2610 step_id: step.id.clone(),
2611 state,
2612 }
2613 .into());
2614 }
2615 if !matches!(
2616 state,
2617 WorkflowStepState::Pending | WorkflowStepState::FailedBeforeDispatch
2618 ) {
2619 return Err(WorkflowResumeError::InvalidState {
2620 step_id: step.id.clone(),
2621 state,
2622 }
2623 .into());
2624 }
2625 for checkpoint_step in checkpoint.steps.iter().skip(next_step_index + 1) {
2626 if checkpoint_step.state != WorkflowStepState::Skipped {
2627 return Err(WorkflowResumeError::InvalidState {
2628 step_id: checkpoint_step.id.clone(),
2629 state: checkpoint_step.state,
2630 }
2631 .into());
2632 }
2633 }
2634 }
2635
2636 Ok(WorkflowResumePlan {
2637 workflow_name: workflow.name.clone(),
2638 workflow_version: workflow.workflow_version.clone(),
2639 next_step_index,
2640 current_revision: current_revision(self),
2641 reconciled: true,
2642 })
2643 }
2644
2645 pub async fn resume_workflow(
2648 &self,
2649 workflow: &WorkflowDefinition,
2650 inputs: &BTreeMap<String, Value>,
2651 checkpoint: &WorkflowCheckpoint,
2652 ) -> BrowserResult<WorkflowRunResult> {
2653 let plan = self
2654 .reconcile_workflow_checkpoint(workflow, checkpoint)
2655 .await?;
2656 if plan.next_step_index >= workflow.steps.len() {
2657 return Err(WorkflowResumeError::CheckpointShape(
2658 "workflow checkpoint is already complete".into(),
2659 )
2660 .into());
2661 }
2662 let mut suffix = workflow.clone();
2663 suffix.steps = workflow.steps[plan.next_step_index..].to_vec();
2664 let mut result = self.run_workflow(&suffix, inputs).await?;
2665 let mut prefix = checkpoint.steps[..plan.next_step_index]
2666 .iter()
2667 .map(checkpoint_step_to_record)
2668 .collect::<Vec<_>>();
2669 prefix.append(&mut result.steps);
2670 result.steps = prefix;
2671 result.trace = WorkflowTrace::from_steps(&result.steps);
2672 result.trace.run_id = Some(result.run_id.clone());
2673 Ok(result)
2674 }
2675}
2676
2677fn checkpoint_step_to_record(step: &WorkflowCheckpointStep) -> WorkflowStepRecord {
2678 let history = if step.history.is_empty() {
2679 vec![
2680 WorkflowStepState::Pending,
2681 WorkflowStepState::Ready,
2682 WorkflowStepState::Preflight,
2683 WorkflowStepState::Resolving,
2684 WorkflowStepState::Dispatched,
2685 WorkflowStepState::EffectObserved,
2686 WorkflowStepState::Verified,
2687 WorkflowStepState::OutputsExtracted,
2688 WorkflowStepState::Committed,
2689 ]
2690 } else {
2691 step.history.clone()
2692 };
2693 WorkflowStepRecord {
2694 id: step.id.clone(),
2695 state: step.state,
2696 history,
2697 attempts: step.attempts,
2698 execution_ids: step.execution_ids.clone(),
2699 dispatch_acknowledged: step.dispatch_acknowledged,
2700 effect_observed: step.effect_observed,
2701 postcondition_verified: step.postcondition_verified,
2702 retry_safe: step.retry_safe,
2703 previous_revision: step.previous_revision,
2704 current_revision: step.current_revision,
2705 branch_decision: step.branch_decision.clone(),
2706 intent_evidence: step.intent_evidence.clone(),
2707 error: None,
2708 }
2709}
2710
2711impl WorkflowCheckpoint {
2712 pub fn to_canonical_json(&self) -> Result<String, WorkflowResumeError> {
2714 self.validate_size()?;
2715 serde_json::to_string(self)
2716 .map_err(|error| WorkflowResumeError::CheckpointShape(error.to_string()))
2717 }
2718
2719 fn validate_size(&self) -> Result<(), WorkflowResumeError> {
2720 if self.schema_version != WORKFLOW_CHECKPOINT_SCHEMA_VERSION {
2721 return Err(WorkflowResumeError::SchemaVersionMismatch {
2722 expected: WORKFLOW_CHECKPOINT_SCHEMA_VERSION,
2723 found: self.schema_version,
2724 });
2725 }
2726 if self.steps.len() > MAX_STEPS {
2727 return Err(WorkflowResumeError::CheckpointShape(
2728 "checkpoint contains too many steps".into(),
2729 ));
2730 }
2731 if self.run_id.len() > 128 {
2732 return Err(WorkflowResumeError::CheckpointShape(
2733 "checkpoint runId is too long".into(),
2734 ));
2735 }
2736 for (index, step) in self.steps.iter().enumerate() {
2737 if step.history.len() > MAX_CHECKPOINT_HISTORY_STATES {
2738 return Err(WorkflowResumeError::CheckpointShape(format!(
2739 "checkpoint step {index} contains too much state history"
2740 )));
2741 }
2742 if !step.history.is_empty() && step.history.last() != Some(&step.state) {
2743 return Err(WorkflowResumeError::CheckpointShape(format!(
2744 "checkpoint step {index} history does not end at its state"
2745 )));
2746 }
2747 if step.execution_ids.len() > MAX_CHECKPOINT_EXECUTION_IDS
2748 || step.execution_ids.iter().any(|id| id.len() > 128)
2749 {
2750 return Err(WorkflowResumeError::CheckpointShape(format!(
2751 "checkpoint step {index} contains too many execution IDs"
2752 )));
2753 }
2754 }
2755 let bytes = serde_json::to_vec(self)
2756 .map_err(|error| WorkflowResumeError::CheckpointShape(error.to_string()))?;
2757 if bytes.len() > MAX_WORKFLOW_CHECKPOINT_BYTES {
2758 return Err(WorkflowResumeError::CheckpointTooLarge);
2759 }
2760 Ok(())
2761 }
2762}
2763
2764fn workflow_definition_hash(
2765 workflow: &WorkflowDefinition,
2766) -> Result<String, WorkflowValidationError> {
2767 let canonical = workflow.to_canonical_json()?;
2768 let digest = Sha256::digest(canonical.as_bytes());
2769 Ok(digest.iter().map(|byte| format!("{byte:02x}")).collect())
2770}
2771
2772fn workflow_target(action: &BatchStep) -> Option<&str> {
2773 match action {
2774 BatchStep::Click { target }
2775 | BatchStep::Check { target }
2776 | BatchStep::Uncheck { target }
2777 | BatchStep::Select { target, .. }
2778 | BatchStep::Clear { target } => Some(target),
2779 BatchStep::Type { target, .. } => target.as_deref(),
2780 _ => None,
2781 }
2782}
2783
2784fn resolve_workflow_intent(
2785 intent: &WorkflowIntentStep,
2786 inputs: &BTreeMap<String, Value>,
2787 path: &str,
2788) -> Result<WorkflowIntentStep, WorkflowValidationError> {
2789 let resolve = |field: &str, value: &str, maximum: usize| {
2790 resolve_input_template(value, inputs, &format!("{path}.{field}"), maximum)
2791 };
2792 let resolve_optional = |field: &str, value: &Option<String>, maximum: usize| {
2793 value
2794 .as_deref()
2795 .map(|value| resolve(field, value, maximum))
2796 .transpose()
2797 };
2798 let mut resolved = intent.clone();
2799 resolved.purpose = resolve_optional("purpose", &intent.purpose, MAX_INTENT_PURPOSE_BYTES)?;
2800 resolved.intent = resolve_optional("intent", &intent.intent, MAX_INTENT_PURPOSE_BYTES * 2)?;
2801 resolved.value = resolve_optional("value", &intent.value, 4_096)?;
2802 resolved.scope.region_id = resolve_optional("scope.regionId", &intent.scope.region_id, 128)?;
2803 resolved.scope.form_label = resolve_optional("scope.formLabel", &intent.scope.form_label, 256)?;
2804 resolved.constraints.role = resolve_optional("constraints.role", &intent.constraints.role, 64)?;
2805 resolved.constraints.name = resolve_optional(
2806 "constraints.name",
2807 &intent.constraints.name,
2808 MAX_TARGET_BYTES,
2809 )?;
2810 resolved.constraints.name_contains = resolve_optional(
2811 "constraints.nameContains",
2812 &intent.constraints.name_contains,
2813 MAX_TARGET_BYTES,
2814 )?;
2815 resolved.constraints.exclude_text = intent
2816 .constraints
2817 .exclude_text
2818 .iter()
2819 .enumerate()
2820 .map(|(index, value)| {
2821 resolve_input_template(
2822 value,
2823 inputs,
2824 &format!("{path}.constraints.excludeText[{index}]"),
2825 MAX_TARGET_BYTES,
2826 )
2827 })
2828 .collect::<Result<_, _>>()?;
2829 resolved.validate(path)?;
2830 Ok(resolved)
2831}
2832
2833fn resolve_batch_step(
2834 action: &BatchStep,
2835 inputs: &BTreeMap<String, Value>,
2836 path: &str,
2837) -> Result<BatchStep, WorkflowValidationError> {
2838 let resolve = |field: &str, value: &str, maximum: usize| {
2839 resolve_input_template(value, inputs, &format!("{path}.{field}"), maximum)
2840 };
2841 Ok(match action {
2842 BatchStep::Navigate { url, timeout_ms } => BatchStep::Navigate {
2843 url: resolve("url", url, MAX_TARGET_BYTES)?,
2844 timeout_ms: *timeout_ms,
2845 },
2846 BatchStep::Click { target } => BatchStep::Click {
2847 target: resolve("target", target, MAX_TARGET_BYTES)?,
2848 },
2849 BatchStep::Type { text, target } => BatchStep::Type {
2850 text: resolve("text", text, MAX_TEXT_BYTES)?,
2851 target: target
2852 .as_deref()
2853 .map(|value| resolve("target", value, MAX_TARGET_BYTES))
2854 .transpose()?,
2855 },
2856 BatchStep::Check { target } => BatchStep::Check {
2857 target: resolve("target", target, MAX_TARGET_BYTES)?,
2858 },
2859 BatchStep::Uncheck { target } => BatchStep::Uncheck {
2860 target: resolve("target", target, MAX_TARGET_BYTES)?,
2861 },
2862 BatchStep::Select { target, value } => BatchStep::Select {
2863 target: resolve("target", target, MAX_TARGET_BYTES)?,
2864 value: resolve("value", value, MAX_TEXT_BYTES)?,
2865 },
2866 BatchStep::Clear { target } => BatchStep::Clear {
2867 target: resolve("target", target, MAX_TARGET_BYTES)?,
2868 },
2869 BatchStep::Wait {
2870 condition,
2871 timeout_ms,
2872 } => BatchStep::Wait {
2873 condition: resolve("condition", condition, MAX_WAIT_CONDITION_BYTES)?,
2874 timeout_ms: *timeout_ms,
2875 },
2876 BatchStep::Scroll { dx, dy } => BatchStep::Scroll { dx: *dx, dy: *dy },
2877 BatchStep::Observe {
2878 include_dom,
2879 include_screenshot,
2880 include_form_values,
2881 } => BatchStep::Observe {
2882 include_dom: *include_dom,
2883 include_screenshot: *include_screenshot,
2884 include_form_values: *include_form_values,
2885 },
2886 BatchStep::Screenshot => BatchStep::Screenshot,
2887 BatchStep::Evaluate { expression } => BatchStep::Evaluate {
2888 expression: resolve("expression", expression, MAX_TEXT_BYTES)?,
2889 },
2890 BatchStep::AcceptDialog => BatchStep::AcceptDialog,
2891 BatchStep::DismissDialog => BatchStep::DismissDialog,
2892 })
2893}
2894
2895fn resolve_input_template(
2896 value: &str,
2897 inputs: &BTreeMap<String, Value>,
2898 path: &str,
2899 maximum: usize,
2900) -> Result<String, WorkflowValidationError> {
2901 let marker = "${inputs.";
2902 let mut resolved = String::with_capacity(value.len());
2903 let mut remainder = value;
2904 while let Some(start) = remainder.find(marker) {
2905 resolved.push_str(&remainder[..start]);
2906 let placeholder = &remainder[start..];
2907 let end = placeholder.find('}').ok_or_else(|| {
2908 WorkflowValidationError::new(path, "input placeholder is missing a closing brace")
2909 })?;
2910 let name = &placeholder[marker.len()..end];
2911 if name.is_empty() || name.contains(['{', '}', '$']) {
2912 return Err(WorkflowValidationError::new(
2913 path,
2914 "input placeholder name is invalid",
2915 ));
2916 }
2917 let input = inputs.get(name).ok_or_else(|| {
2918 WorkflowValidationError::new(path, format!("input {name:?} is missing or not declared"))
2919 })?;
2920 let text = match input {
2921 Value::String(text) => text.clone(),
2922 Value::Bool(value) => value.to_string(),
2923 Value::Number(value) => value.to_string(),
2924 _ => {
2925 return Err(WorkflowValidationError::new(
2926 path,
2927 format!("input {name:?} cannot be inserted into an action string"),
2928 ));
2929 }
2930 };
2931 resolved.push_str(&text);
2932 remainder = &placeholder[end + 1..];
2933 }
2934 resolved.push_str(remainder);
2935 validate_bytes(path, &resolved, 0, maximum)?;
2936 Ok(resolved)
2937}
2938
2939async fn extract_workflow_outputs(
2940 session: &super::BrowserSession,
2941 workflow: &WorkflowDefinition,
2942) -> BrowserResult<BTreeMap<String, WorkflowOutput>> {
2943 let page = if workflow
2944 .outputs
2945 .values()
2946 .any(|output| !matches!(output.source, WorkflowOutputSource::VisibleText))
2947 {
2948 Some(session.page_info().await?)
2949 } else {
2950 None
2951 };
2952 let visible_text = if workflow
2953 .outputs
2954 .values()
2955 .any(|output| matches!(output.source, WorkflowOutputSource::VisibleText))
2956 {
2957 Some(session.text().await?)
2958 } else {
2959 None
2960 };
2961 let mut outputs = BTreeMap::new();
2962 let mut extracted_bytes = 0usize;
2963 let revision = current_revision(session);
2964 for (name, declaration) in &workflow.outputs {
2965 let text = match declaration.source {
2966 WorkflowOutputSource::PageUrl => page.as_ref().map(|page| page.url.as_str()),
2967 WorkflowOutputSource::PageTitle => page.as_ref().map(|page| page.title.as_str()),
2968 WorkflowOutputSource::VisibleText => visible_text.as_deref(),
2969 }
2970 .ok_or_else(|| format!("output {name:?} has no extraction source"))?;
2971 extracted_bytes = extracted_bytes.saturating_add(text.len());
2972 if extracted_bytes > workflow.budgets.max_extracted_bytes {
2973 return Err(format!(
2974 "outputs exceed maxExtractedBytes {}",
2975 workflow.budgets.max_extracted_bytes
2976 )
2977 .into());
2978 }
2979 let value = typed_output_value(name, declaration.value_type, text)?;
2980 let redacted = declaration.sensitive;
2981 outputs.insert(
2982 name.clone(),
2983 WorkflowOutput {
2984 value_type: declaration.value_type,
2985 value: if redacted { Value::Null } else { value },
2986 redacted,
2987 evidence: WorkflowOutputEvidence {
2988 source: declaration.source,
2989 revision,
2990 },
2991 },
2992 );
2993 }
2994 Ok(outputs)
2995}
2996
2997fn typed_output_value(
2998 name: &str,
2999 value_type: WorkflowValueType,
3000 text: &str,
3001) -> BrowserResult<Value> {
3002 let trimmed = text.trim();
3003 match value_type {
3004 WorkflowValueType::String => Ok(Value::String(text.to_string())),
3005 WorkflowValueType::Url => {
3006 if Url::parse(trimmed).is_err() {
3007 return Err(format!("output {name:?} is not a valid URL").into());
3008 }
3009 Ok(Value::String(text.to_string()))
3010 }
3011 WorkflowValueType::Integer => trimmed
3012 .parse::<i64>()
3013 .map(Value::from)
3014 .map_err(|_| format!("output {name:?} cannot be parsed as an integer").into()),
3015 WorkflowValueType::Number => {
3016 let number = trimmed
3017 .parse::<f64>()
3018 .map_err(|_| format!("output {name:?} cannot be parsed as a number"))?;
3019 serde_json::Number::from_f64(number)
3020 .map(Value::Number)
3021 .ok_or_else(|| format!("output {name:?} is not a finite number").into())
3022 }
3023 WorkflowValueType::Boolean => match trimmed {
3024 "true" => Ok(Value::Bool(true)),
3025 "false" => Ok(Value::Bool(false)),
3026 _ => Err(format!("output {name:?} cannot be parsed as a boolean").into()),
3027 },
3028 }
3029}
3030
3031fn current_revision(session: &super::BrowserSession) -> u64 {
3032 session
3033 .page_revision
3034 .load(std::sync::atomic::Ordering::Relaxed)
3035}
3036
3037fn workflow_budget_expired(started: Instant, budget: Duration) -> bool {
3038 started.elapsed() >= budget
3039}
3040
3041fn workflow_budget_remaining(started: Instant, budget: Duration) -> Duration {
3042 budget.saturating_sub(started.elapsed())
3043}
3044
3045fn can_retry_before_dispatch(
3046 transaction: WorkflowTransactionClass,
3047 dispatch_observed: bool,
3048 attempt_number: u32,
3049 max_retries: u32,
3050) -> bool {
3051 !dispatch_observed && attempt_number <= max_retries && transaction.permits_pre_dispatch_retry()
3052}
3053
3054fn skip_remaining(records: &mut [WorkflowStepRecord], start: usize) {
3055 for record in records.iter_mut().skip(start) {
3056 if record.state == WorkflowStepState::Pending {
3057 let _ = record.transition(WorkflowStepState::Skipped);
3058 }
3059 }
3060}
3061
3062fn commit_workflow_effect_marker(record: &mut WorkflowStepRecord, revision: u64) {
3063 if record.state == WorkflowStepState::Ready {
3064 let _ = record.transition(WorkflowStepState::Preflight);
3065 }
3066 if record.state == WorkflowStepState::Preflight {
3067 let _ = record.transition(WorkflowStepState::EffectObserved);
3068 }
3069 record.effect_observed = true;
3070 record.postcondition_verified = true;
3071 record.current_revision = Some(revision);
3072 let _ = record.transition(WorkflowStepState::Verified);
3073 let _ = record.transition(WorkflowStepState::OutputsExtracted);
3074 let _ = record.transition(WorkflowStepState::Committed);
3075}
3076
3077fn state_name(state: WorkflowStepState) -> &'static str {
3078 match state {
3079 WorkflowStepState::Pending => "pending",
3080 WorkflowStepState::Ready => "ready",
3081 WorkflowStepState::Preflight => "preflight",
3082 WorkflowStepState::Resolving => "resolving",
3083 WorkflowStepState::NotDispatched => "not_dispatched",
3084 WorkflowStepState::Dispatched => "dispatched",
3085 WorkflowStepState::EffectObserved => "effect_observed",
3086 WorkflowStepState::Verified => "verified",
3087 WorkflowStepState::OutputsExtracted => "outputs_extracted",
3088 WorkflowStepState::Committed => "committed",
3089 WorkflowStepState::FailedBeforeDispatch => "failed_before_dispatch",
3090 WorkflowStepState::FailedAfterDispatch => "failed_after_dispatch",
3091 WorkflowStepState::Indeterminate => "indeterminate",
3092 WorkflowStepState::Skipped => "skipped",
3093 }
3094}
3095
3096fn bound_workflow_text(value: &str, max_bytes: usize) -> String {
3097 if value.len() <= max_bytes {
3098 return value.to_string();
3099 }
3100 let mut end = max_bytes.saturating_sub(13);
3101 while end > 0 && !value.is_char_boundary(end) {
3102 end -= 1;
3103 }
3104 format!("{}\n[truncated]", &value[..end])
3105}
3106
3107#[derive(Debug, Clone, Serialize, Deserialize)]
3109#[serde(rename_all = "camelCase")]
3110pub struct WorkflowOutputDeclaration {
3111 #[serde(alias = "type")]
3112 pub value_type: WorkflowValueType,
3113 pub source: WorkflowOutputSource,
3114 #[serde(default)]
3115 pub required: bool,
3116 #[serde(default)]
3117 pub sensitive: bool,
3118}
3119
3120impl WorkflowOutputDeclaration {
3121 fn validate(&self, path: &str) -> Result<(), WorkflowValidationError> {
3122 let compatible = match self.source {
3123 WorkflowOutputSource::PageUrl => {
3124 matches!(
3125 self.value_type,
3126 WorkflowValueType::String | WorkflowValueType::Url
3127 )
3128 }
3129 WorkflowOutputSource::PageTitle | WorkflowOutputSource::VisibleText => true,
3130 };
3131 if !compatible {
3132 return Err(WorkflowValidationError::new(
3133 format!("{path}.valueType"),
3134 format!(
3135 "{} output source requires a compatible value type",
3136 self.source
3137 ),
3138 ));
3139 }
3140 Ok(())
3141 }
3142}
3143
3144#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
3146#[serde(rename_all = "snake_case")]
3147pub enum WorkflowOutputSource {
3148 PageUrl,
3149 PageTitle,
3150 VisibleText,
3151}
3152
3153impl fmt::Display for WorkflowOutputSource {
3154 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
3155 formatter.write_str(match self {
3156 Self::PageUrl => "page_url",
3157 Self::PageTitle => "page_title",
3158 Self::VisibleText => "visible_text",
3159 })
3160 }
3161}
3162
3163#[derive(Debug, Clone, Serialize, Deserialize)]
3165#[serde(rename_all = "camelCase")]
3166pub struct WorkflowOutput {
3167 pub value_type: WorkflowValueType,
3168 pub value: Value,
3169 #[serde(default, skip_serializing_if = "is_false")]
3170 pub redacted: bool,
3171 pub evidence: WorkflowOutputEvidence,
3172}
3173
3174fn is_false(value: &bool) -> bool {
3175 !*value
3176}
3177
3178#[derive(Debug, Clone, Serialize, Deserialize)]
3180#[serde(rename_all = "camelCase")]
3181pub struct WorkflowOutputEvidence {
3182 pub source: WorkflowOutputSource,
3183 pub revision: u64,
3184}
3185
3186#[derive(Debug, Clone, Serialize)]
3188pub struct WorkflowValidationError {
3189 pub path: String,
3190 pub reason: String,
3191}
3192
3193impl WorkflowValidationError {
3194 pub(crate) fn new(path: impl Into<String>, reason: impl Into<String>) -> Self {
3195 Self {
3196 path: path.into(),
3197 reason: reason.into(),
3198 }
3199 }
3200}
3201
3202impl fmt::Display for WorkflowValidationError {
3203 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
3204 write!(formatter, "{}: {}", self.path, self.reason)
3205 }
3206}
3207
3208impl std::error::Error for WorkflowValidationError {}
3209
3210fn require_object_fields(value: &Value, fields: &[&str]) -> Result<(), WorkflowValidationError> {
3211 let object = value.as_object().ok_or_else(|| {
3212 WorkflowValidationError::new("$", "workflow definition must be a JSON object")
3213 })?;
3214 for field in fields {
3215 if !object.contains_key(*field) {
3216 return Err(WorkflowValidationError::new(
3217 format!("$.{field}"),
3218 "required field is missing",
3219 ));
3220 }
3221 }
3222 Ok(())
3223}
3224
3225fn reject_unknown_fields(
3226 value: &Value,
3227 path: &str,
3228 allowed: &[&str],
3229) -> Result<(), WorkflowValidationError> {
3230 let object = value
3231 .as_object()
3232 .ok_or_else(|| WorkflowValidationError::new(path, "expected a JSON object"))?;
3233 if let Some(field) = object
3234 .keys()
3235 .find(|field| !allowed.contains(&field.as_str()))
3236 {
3237 return Err(WorkflowValidationError::new(
3238 format!("{path}.{field}"),
3239 "unknown workflow field",
3240 ));
3241 }
3242 Ok(())
3243}
3244
3245fn reject_predicate_fields(value: &Value, path: &str) -> Result<(), WorkflowValidationError> {
3246 let object = value
3247 .as_object()
3248 .ok_or_else(|| WorkflowValidationError::new(path, "expected a predicate object"))?;
3249 let allowed = [
3250 "urlEquals",
3251 "titleContains",
3252 "visible",
3253 "textContains",
3254 "popupOpened",
3255 "dialogOpen",
3256 "downloadStarted",
3257 "revisionEquals",
3258 "all",
3259 "any",
3260 "not",
3261 ];
3262 if object.len() != 1 {
3263 return Err(WorkflowValidationError::new(
3264 path,
3265 "predicate must contain exactly one recognized field",
3266 ));
3267 }
3268 let Some((field, nested)) = object.iter().next() else {
3269 return Err(WorkflowValidationError::new(
3270 path,
3271 "predicate must not be empty",
3272 ));
3273 };
3274 if !allowed.contains(&field.as_str()) {
3275 return Err(WorkflowValidationError::new(
3276 format!("{path}.{field}"),
3277 "unknown predicate field",
3278 ));
3279 }
3280 match field.as_str() {
3281 "all" | "any" => {
3282 let predicates = nested.as_array().ok_or_else(|| {
3283 WorkflowValidationError::new(format!("{path}.{field}"), "must be an array")
3284 })?;
3285 for (index, predicate) in predicates.iter().enumerate() {
3286 reject_predicate_fields(predicate, &format!("{path}.{field}[{index}]"))?;
3287 }
3288 }
3289 "not" => reject_predicate_fields(nested, &format!("{path}.not"))?,
3290 _ => {}
3291 }
3292 Ok(())
3293}
3294
3295fn validate_name(path: &str, value: &str) -> Result<(), WorkflowValidationError> {
3296 validate_bytes(path, value, 1, MAX_NAME_BYTES)?;
3297 if value.trim() != value {
3298 return Err(WorkflowValidationError::new(
3299 path,
3300 "must not have leading or trailing whitespace",
3301 ));
3302 }
3303 Ok(())
3304}
3305
3306fn validate_map_key(scope: &str, key: &str) -> Result<(), WorkflowValidationError> {
3307 validate_name(&format!("{scope}.{key}"), key)
3308}
3309
3310fn validate_bytes(
3311 path: &str,
3312 value: &str,
3313 minimum: usize,
3314 maximum: usize,
3315) -> Result<(), WorkflowValidationError> {
3316 if value.len() < minimum || value.len() > maximum {
3317 return Err(WorkflowValidationError::new(
3318 path,
3319 format!("must be {minimum}..={maximum} UTF-8 bytes"),
3320 ));
3321 }
3322 Ok(())
3323}
3324
3325fn validate_target(path: &str, target: &str) -> Result<(), WorkflowValidationError> {
3326 validate_bytes(path, target, 1, MAX_TARGET_BYTES)
3327}
3328
3329fn validate_predicate(
3330 predicate: &VerificationPredicate,
3331 path: &str,
3332) -> Result<(), WorkflowValidationError> {
3333 predicate
3334 .validate(0)
3335 .map_err(|error| WorkflowValidationError::new(path, error.to_string()))
3336}
3337
3338fn validate_batch_step(action: &BatchStep, path: &str) -> Result<(), WorkflowValidationError> {
3339 match action {
3340 BatchStep::Navigate { url, timeout_ms } => {
3341 if Url::parse(url).is_err() {
3342 return Err(WorkflowValidationError::new(
3343 format!("{path}.url"),
3344 "must be an absolute URL",
3345 ));
3346 }
3347 if *timeout_ms == 0 || *timeout_ms > MAX_DURATION_MS {
3348 return Err(WorkflowValidationError::new(
3349 format!("{path}.timeoutMs"),
3350 format!("must be 1..={MAX_DURATION_MS}"),
3351 ));
3352 }
3353 }
3354 BatchStep::Click { target }
3355 | BatchStep::Check { target }
3356 | BatchStep::Uncheck { target }
3357 | BatchStep::Clear { target } => validate_target(&format!("{path}.target"), target)?,
3358 BatchStep::Type { text, target } => {
3359 validate_bytes(&format!("{path}.text"), text, 0, MAX_TEXT_BYTES)?;
3360 if let Some(target) = target {
3361 validate_target(&format!("{path}.target"), target)?;
3362 }
3363 }
3364 BatchStep::Select { target, value } => {
3365 validate_target(&format!("{path}.target"), target)?;
3366 validate_bytes(&format!("{path}.value"), value, 1, MAX_TEXT_BYTES)?;
3367 }
3368 BatchStep::Scroll { dx, dy } => {
3369 if !dx.is_finite()
3370 || !dy.is_finite()
3371 || dx.abs() > 1_000_000.0
3372 || dy.abs() > 1_000_000.0
3373 {
3374 return Err(WorkflowValidationError::new(
3375 path,
3376 "scroll deltas must be finite and bounded",
3377 ));
3378 }
3379 }
3380 BatchStep::Wait {
3381 condition,
3382 timeout_ms,
3383 } => {
3384 validate_bytes(
3385 &format!("{path}.condition"),
3386 condition,
3387 1,
3388 MAX_WAIT_CONDITION_BYTES,
3389 )?;
3390 if *timeout_ms == 0 || *timeout_ms > MAX_DURATION_MS {
3391 return Err(WorkflowValidationError::new(
3392 format!("{path}.timeoutMs"),
3393 format!("must be 1..={MAX_DURATION_MS}"),
3394 ));
3395 }
3396 }
3397 BatchStep::Observe { .. }
3398 | BatchStep::Screenshot
3399 | BatchStep::AcceptDialog
3400 | BatchStep::DismissDialog => {}
3401 BatchStep::Evaluate { .. } => {
3402 return Err(WorkflowValidationError::new(
3403 path,
3404 "evaluate is not permitted in a declarative workflow",
3405 ));
3406 }
3407 }
3408 Ok(())
3409}
3410
3411#[cfg(test)]
3412mod tests {
3413 use super::*;
3414 use crate::browser::session::{
3415 FingerprintInvalidation, SemanticIntentCandidate, SemanticIntentPurpose,
3416 SemanticRegionKind, SemanticTargetFingerprint,
3417 };
3418 use serde_json::json;
3419
3420 fn definition() -> WorkflowDefinition {
3421 WorkflowDefinition {
3422 schema_version: WORKFLOW_SCHEMA_VERSION,
3423 name: "example".into(),
3424 workflow_version: "1.0.0".into(),
3425 description: None,
3426 inputs: BTreeMap::from([(
3427 "url".into(),
3428 WorkflowInput {
3429 value_type: WorkflowValueType::Url,
3430 required: true,
3431 max_length: Some(2_048),
3432 sensitive: None,
3433 },
3434 )]),
3435 budgets: WorkflowBudgets {
3436 max_steps: 2,
3437 max_duration_ms: 30_000,
3438 max_retries: 1,
3439 max_extracted_bytes: 4_096,
3440 },
3441 preconditions: vec![],
3442 steps: vec![WorkflowStep {
3443 id: "open".into(),
3444 action: BatchStep::Navigate {
3445 url: "https://example.com".into(),
3446 timeout_ms: 20_000,
3447 },
3448 intent: None,
3449 when: None,
3450 expect: Some(VerificationPredicate::TitleContains {
3451 value: "Example".into(),
3452 }),
3453 before_retry: None,
3454 transaction: WorkflowTransactionClass::ReadOnly,
3455 idempotency_key: None,
3456 max_retries: 0,
3457 repeat: 1,
3458 }],
3459 terminal_condition: VerificationPredicate::UrlEquals {
3460 value: "https://example.com/".into(),
3461 },
3462 outputs: BTreeMap::new(),
3463 }
3464 }
3465
3466 #[test]
3467 fn canonical_json_is_stable_and_uses_camel_case() {
3468 let workflow = definition();
3469 let first = workflow.to_canonical_json().unwrap();
3470 let second = workflow.to_canonical_json().unwrap();
3471 assert_eq!(first, second);
3472 assert!(first.contains("\"schemaVersion\":1"));
3473 assert!(first.contains("\"timeoutMs\":20000"));
3474 assert!(first.contains("\"workflowVersion\":\"1.0.0\""));
3475 }
3476
3477 #[test]
3478 fn before_retry_marker_is_canonical_and_validated() {
3479 let mut workflow = definition();
3480 workflow.steps[0].before_retry = Some(VerificationPredicate::TitleContains {
3481 value: "Already saved".into(),
3482 });
3483 let json = workflow.to_canonical_json().unwrap();
3484 assert!(json.contains("\"beforeRetry\":{"));
3485 let parsed = WorkflowDefinition::from_json(&json).unwrap();
3486 assert!(parsed.steps[0].before_retry.is_some());
3487 }
3488
3489 #[test]
3490 fn type_alias_is_accepted_but_canonical_json_uses_value_type() {
3491 let mut value = serde_json::to_value(definition()).unwrap();
3492 let input = value["inputs"]["url"].as_object_mut().unwrap();
3493 let value_type = input.remove("valueType").unwrap();
3494 input.insert("type".into(), value_type);
3495
3496 let parsed = WorkflowDefinition::from_value(value).unwrap();
3497 let canonical = parsed.to_canonical_json().unwrap();
3498 assert!(canonical.contains("\"valueType\":\"url\""));
3499 assert!(!canonical.contains("\"type\":\"url\""));
3500 }
3501
3502 #[test]
3503 fn effect_marker_commit_follows_evidence_states_without_dispatch() {
3504 let mut record = WorkflowStepRecord::new("save");
3505 record.transition(WorkflowStepState::Ready).unwrap();
3506 commit_workflow_effect_marker(&mut record, 9);
3507 assert_eq!(record.state, WorkflowStepState::Committed);
3508 assert!(!record.dispatch_acknowledged);
3509 assert!(record.effect_observed);
3510 assert!(record.postcondition_verified);
3511 assert_eq!(record.current_revision, Some(9));
3512 }
3513
3514 #[test]
3515 fn from_json_reports_missing_top_level_path() {
3516 let mut value = serde_json::to_value(definition()).unwrap();
3517 value.as_object_mut().unwrap().remove("steps");
3518 let error = WorkflowDefinition::from_value(value).unwrap_err();
3519 assert_eq!(error.path, "$.steps");
3520 }
3521
3522 #[test]
3523 fn unknown_definition_fields_fail_before_deserialization() {
3524 let mut value = serde_json::to_value(definition()).unwrap();
3525 value
3526 .as_object_mut()
3527 .unwrap()
3528 .insert("futureField".into(), json!(true));
3529 let error = WorkflowDefinition::from_value(value).unwrap_err();
3530 assert_eq!(error.path, "$.futureField");
3531 }
3532
3533 #[test]
3534 fn semantic_intent_steps_round_trip_with_bounded_defaults() {
3535 let mut value = serde_json::to_value(definition()).unwrap();
3536 value["steps"][0] = json!({
3537 "id": "continue",
3538 "intent": {
3539 "action": "click",
3540 "purpose": "continueCheckout",
3541 "scope": {"regionKind": "checkoutSummary"},
3542 "resolutionPolicy": "requireUniqueHighConfidence"
3543 },
3544 "transaction": "idempotent"
3545 });
3546 let workflow = WorkflowDefinition::from_value(value).unwrap();
3547 let intent = workflow.steps[0].intent.as_ref().unwrap();
3548 assert_eq!(intent.action, SemanticIntentAction::Click);
3549 assert_eq!(intent.purpose.as_deref(), Some("continueCheckout"));
3550 assert_eq!(
3551 intent
3552 .execution_request("steps[0].intent")
3553 .unwrap()
3554 .request
3555 .intent,
3556 "continue checkout"
3557 );
3558 let canonical = workflow.to_canonical_json().unwrap();
3559 assert!(canonical.contains("\"resolutionPolicy\":\"requireUniqueHighConfidence\""));
3560 assert!(!canonical.contains("\"action\":\"navigate\""));
3561 }
3562
3563 #[test]
3564 fn semantic_intent_steps_reject_ambiguous_shape_and_missing_values() {
3565 let mut value = serde_json::to_value(definition()).unwrap();
3566 value["steps"][0] = json!({
3567 "id": "bad",
3568 "intent": {
3569 "action": "type",
3570 "purpose": "enterSearch",
3571 "intent": "enter search",
3572 "resolutionPolicy": "requireUniqueHighConfidence"
3573 },
3574 "transaction": "idempotent"
3575 });
3576 let error = WorkflowDefinition::from_value(value).unwrap_err();
3577 assert_eq!(error.path, "steps[0].intent.purpose");
3578
3579 let mut value = serde_json::to_value(definition()).unwrap();
3580 value["steps"][0] = json!({
3581 "id": "bad",
3582 "intent": {
3583 "action": "type",
3584 "purpose": "enterSearch",
3585 "resolutionPolicy": "requireUniqueHighConfidence"
3586 },
3587 "transaction": "idempotent"
3588 });
3589 let error = WorkflowDefinition::from_value(value).unwrap_err();
3590 assert_eq!(error.path, "steps[0].intent.value");
3591 }
3592
3593 #[test]
3594 fn unknown_predicate_fields_fail_with_their_json_path() {
3595 let mut value = serde_json::to_value(definition()).unwrap();
3596 value["terminalCondition"] = json!({"titleContains": "Example", "future": true});
3597 let error = WorkflowDefinition::from_value(value).unwrap_err();
3598 assert_eq!(error.path, "terminalCondition");
3599 }
3600
3601 #[test]
3602 fn duplicate_step_ids_are_rejected() {
3603 let mut workflow = definition();
3604 workflow.steps.push(WorkflowStep {
3605 id: "open".into(),
3606 action: BatchStep::Screenshot,
3607 intent: None,
3608 when: None,
3609 expect: None,
3610 before_retry: None,
3611 transaction: WorkflowTransactionClass::ReadOnly,
3612 idempotency_key: None,
3613 max_retries: 0,
3614 repeat: 1,
3615 });
3616 let error = workflow.validate().unwrap_err();
3617 assert_eq!(error.path, "steps[1].id");
3618 }
3619
3620 #[test]
3621 fn invalid_budget_reports_exact_path() {
3622 let mut workflow = definition();
3623 workflow.budgets.max_steps = 0;
3624 let error = workflow.validate().unwrap_err();
3625 assert_eq!(error.path, "budgets.maxSteps");
3626 }
3627
3628 #[test]
3629 fn conditional_idempotency_requires_a_key() {
3630 let mut workflow = definition();
3631 workflow.steps[0].transaction = WorkflowTransactionClass::ConditionallyIdempotent;
3632 let error = workflow.validate().unwrap_err();
3633 assert_eq!(error.path, "steps[0].idempotencyKey");
3634 }
3635
3636 #[test]
3637 fn duplicate_idempotency_keys_are_rejected() {
3638 let mut workflow = definition();
3639 workflow.steps[0].transaction = WorkflowTransactionClass::ConditionallyIdempotent;
3640 workflow.steps[0].idempotency_key = Some("save-once".into());
3641 workflow.steps.push(WorkflowStep {
3642 id: "second".into(),
3643 action: BatchStep::Screenshot,
3644 intent: None,
3645 when: None,
3646 expect: None,
3647 before_retry: None,
3648 transaction: WorkflowTransactionClass::ConditionallyIdempotent,
3649 idempotency_key: Some("save-once".into()),
3650 max_retries: 0,
3651 repeat: 1,
3652 });
3653 workflow.budgets.max_steps = 2;
3654 let error = workflow.validate().unwrap_err();
3655 assert_eq!(error.path, "steps[1].idempotencyKey");
3656 }
3657
3658 #[test]
3659 fn unknown_steps_cannot_request_automatic_retries() {
3660 let mut workflow = definition();
3661 workflow.steps[0].transaction = WorkflowTransactionClass::Unknown;
3662 workflow.budgets.max_retries = 1;
3663 workflow.steps[0].max_retries = 1;
3664 let error = workflow.validate().unwrap_err();
3665 assert_eq!(error.path, "steps[0].maxRetries");
3666 }
3667
3668 #[test]
3669 fn bounded_repetition_requires_retry_safe_class() {
3670 let mut workflow = definition();
3671 workflow.budgets.max_steps = 2;
3672 workflow.steps[0].repeat = 2;
3673 workflow.steps[0].transaction = WorkflowTransactionClass::Unknown;
3674 let error = workflow.validate().unwrap_err();
3675 assert_eq!(error.path, "steps[0].repeat");
3676 }
3677
3678 #[test]
3679 fn retry_policy_never_replays_after_dispatch() {
3680 assert!(can_retry_before_dispatch(
3681 WorkflowTransactionClass::Idempotent,
3682 false,
3683 1,
3684 1
3685 ));
3686 assert!(!can_retry_before_dispatch(
3687 WorkflowTransactionClass::NonIdempotent,
3688 false,
3689 1,
3690 1
3691 ));
3692 assert!(!can_retry_before_dispatch(
3693 WorkflowTransactionClass::Idempotent,
3694 true,
3695 1,
3696 1
3697 ));
3698 }
3699
3700 #[test]
3701 fn workflow_checkpoint_is_deterministic_and_redacted() {
3702 let checkpoint = WorkflowCheckpoint {
3703 schema_version: WORKFLOW_CHECKPOINT_SCHEMA_VERSION,
3704 run_id: "run_test".into(),
3705 workflow_name: "example".into(),
3706 workflow_version: "1.0.0".into(),
3707 definition_hash: "a".repeat(64),
3708 status: WorkflowRunStatus::Failed,
3709 next_step_index: 1,
3710 steps: vec![
3711 WorkflowCheckpointStep {
3712 id: "open".into(),
3713 state: WorkflowStepState::Committed,
3714 attempts: 1,
3715 history: Vec::new(),
3716 execution_ids: Vec::new(),
3717 dispatch_acknowledged: false,
3718 effect_observed: false,
3719 postcondition_verified: false,
3720 retry_safe: false,
3721 previous_revision: None,
3722 current_revision: None,
3723 branch_decision: None,
3724 intent_evidence: None,
3725 },
3726 WorkflowCheckpointStep {
3727 id: "save".into(),
3728 state: WorkflowStepState::FailedBeforeDispatch,
3729 attempts: 1,
3730 history: Vec::new(),
3731 execution_ids: Vec::new(),
3732 dispatch_acknowledged: false,
3733 effect_observed: false,
3734 postcondition_verified: false,
3735 retry_safe: false,
3736 previous_revision: None,
3737 current_revision: None,
3738 branch_decision: None,
3739 intent_evidence: None,
3740 },
3741 ],
3742 page: WorkflowCheckpointPage {
3743 target_id: "target".into(),
3744 frame_id: "frame".into(),
3745 url: "https://example.com".into(),
3746 title: "Example".into(),
3747 revision: 3,
3748 },
3749 };
3750 let first = checkpoint.to_canonical_json().unwrap();
3751 let second = checkpoint.to_canonical_json().unwrap();
3752 assert_eq!(first, second);
3753 assert!(!first.contains("password"));
3754 assert_eq!(
3755 crate::BrowserSession::parse_workflow_checkpoint(&first)
3756 .unwrap()
3757 .next_step_index,
3758 1
3759 );
3760 }
3761
3762 #[test]
3763 fn minimal_workflow_fixture_round_trips() {
3764 let workflow = WorkflowDefinition::from_json(include_str!(
3765 "../../../tests/fixtures/workflow-minimal.json"
3766 ))
3767 .unwrap();
3768 assert_eq!(workflow.name, "open-example");
3769 assert!(workflow.preconditions.is_empty());
3770 assert!(workflow.to_canonical_json().is_ok());
3771 }
3772
3773 #[test]
3774 fn evaluate_is_rejected_before_execution() {
3775 let mut workflow = definition();
3776 workflow.steps[0].action = BatchStep::Evaluate {
3777 expression: "document.title".into(),
3778 };
3779 let error = workflow.validate().unwrap_err();
3780 assert!(error.reason.contains("not permitted"));
3781 }
3782
3783 #[test]
3784 fn inputs_are_type_checked_and_bounded() {
3785 let workflow = definition();
3786 let values = BTreeMap::from([("url".into(), json!("https://example.com"))]);
3787 workflow.validate_inputs(&values).unwrap();
3788
3789 let bad_values = BTreeMap::from([("url".into(), json!("not a url"))]);
3790 let error = workflow.validate_inputs(&bad_values).unwrap_err();
3791 assert_eq!(error.path, "inputs.url");
3792 }
3793
3794 #[test]
3795 fn resolve_actions_substitutes_bounded_inputs() {
3796 let mut workflow = definition();
3797 workflow.steps[0].action = BatchStep::Navigate {
3798 url: "${inputs.url}".into(),
3799 timeout_ms: 20_000,
3800 };
3801 let values = BTreeMap::from([("url".into(), json!("https://docs.example.com"))]);
3802 let steps = workflow.resolve_actions(&values).unwrap();
3803 let BatchStep::Navigate { url, .. } = &steps[0].action else {
3804 panic!("expected navigate action");
3805 };
3806 assert_eq!(url, "https://docs.example.com");
3807 }
3808
3809 #[test]
3810 fn resolve_actions_rejects_unknown_placeholders_before_dispatch() {
3811 let mut workflow = definition();
3812 workflow.steps[0].action = BatchStep::Click {
3813 target: "name=${inputs.missing}".into(),
3814 };
3815 let values = BTreeMap::from([("url".into(), json!("https://example.com"))]);
3816 let error = workflow.resolve_actions(&values).unwrap_err();
3817 assert_eq!(error.path, "steps[0].action.target");
3818 }
3819
3820 #[test]
3821 fn typed_output_values_require_strict_scalar_conversions() {
3822 assert_eq!(
3823 typed_output_value("count", WorkflowValueType::Integer, " 42 ").unwrap(),
3824 json!(42)
3825 );
3826 assert_eq!(
3827 typed_output_value("ready", WorkflowValueType::Boolean, "false").unwrap(),
3828 json!(false)
3829 );
3830 assert!(typed_output_value("count", WorkflowValueType::Integer, "4.2").is_err());
3831 assert!(typed_output_value("ready", WorkflowValueType::Boolean, "yes").is_err());
3832 }
3833
3834 #[test]
3835 fn sensitive_output_serialization_contains_no_literal_value() {
3836 let output = WorkflowOutput {
3837 value_type: WorkflowValueType::String,
3838 value: Value::Null,
3839 redacted: true,
3840 evidence: WorkflowOutputEvidence {
3841 source: WorkflowOutputSource::VisibleText,
3842 revision: 4,
3843 },
3844 };
3845 let serialized = serde_json::to_string(&output).unwrap();
3846 assert!(serialized.contains("\"redacted\":true"));
3847 assert!(!serialized.contains("secret"));
3848 }
3849
3850 #[test]
3851 fn recorder_keeps_semantic_targets_and_redacts_typed_values() {
3852 let mut recorder = WorkflowRecorder::new("checkout", "1.0.0");
3853 recorder
3854 .record_click("continue", "button", "Continue", None)
3855 .unwrap();
3856 recorder
3857 .record_type_input("email", "textbox", "Email", "email")
3858 .unwrap();
3859 let draft = recorder.draft();
3860 let BatchStep::Click { target } = &draft.steps[0].action else {
3861 panic!("expected semantic click draft");
3862 };
3863 assert_eq!(target, "role=button;name=Continue");
3864 let serialized = serde_json::to_string(draft).unwrap();
3865 assert!(!serialized.contains("password-value"));
3866 assert!(serialized.contains("${inputs.email}"));
3867 assert!(draft.steps[1].review_required);
3868 let inferred = recorder.inferred_inputs();
3869 assert_eq!(inferred["email"].value_type, WorkflowValueType::String);
3870 assert_eq!(inferred["email"].sensitive, None);
3871 }
3872
3873 #[test]
3874 fn recorder_captures_semantic_evidence_without_replay_handles_or_query_values() {
3875 let request = SemanticIntentRequest {
3876 schema_version: INTENT_RESOLUTION_SCHEMA_VERSION,
3877 intent: "submit order".into(),
3878 action: SemanticIntentAction::Click,
3879 scope: IntentScope::default(),
3880 constraints: IntentConstraints::default(),
3881 resolution_policy: SemanticResolutionPolicy::RequireUniqueHighConfidence,
3882 expected_revision: None,
3883 };
3884 let route = SemanticRouteIdentity {
3885 target_id: "target-7".into(),
3886 frame_id: "frame-2".into(),
3887 url: "https://shop.example/orders?token=secret#confirmation".into(),
3888 };
3889 let result = SemanticIntentResult {
3890 schema_version: INTENT_RESOLUTION_SCHEMA_VERSION,
3891 intent: request.intent.clone(),
3892 action: request.action,
3893 normalized_intent: "submit order".into(),
3894 resolution: SemanticResolution::UniqueHighConfidence,
3895 policy_decision: IntentPolicyDecision::Allowed,
3896 route: Some(route.clone()),
3897 revision: Some(7),
3898 candidates: vec![SemanticIntentCandidate {
3899 id: "candidate-1".into(),
3900 reference: "r7:backend-secret".into(),
3901 role: "button".into(),
3902 name: "Submit order".into(),
3903 input_type: None,
3904 region_id: Some("checkout-region".into()),
3905 region_kind: Some(SemanticRegionKind::CheckoutSummary),
3906 confidence: IntentConfidence::High,
3907 evidence: Vec::new(),
3908 fingerprint: Some(SemanticTargetFingerprint {
3909 revision: 7,
3910 route: route.clone(),
3911 role: "button".into(),
3912 name: "Submit order".into(),
3913 input_type: None,
3914 region_id: Some("checkout-region".into()),
3915 region_kind: Some(SemanticRegionKind::CheckoutSummary),
3916 purpose: SemanticIntentPurpose::Submit,
3917 invalidated_by: vec![FingerprintInvalidation::Revision],
3918 }),
3919 }],
3920 excluded_candidates: Vec::new(),
3921 excluded_count: 0,
3922 selected_candidate: Some("candidate-1".into()),
3923 suggested_constraints: Vec::new(),
3924 reason: None,
3925 };
3926 let mut recorder = WorkflowRecorder::new("checkout", "1.0.0");
3927 recorder
3928 .record_semantic_intent(
3929 "submit",
3930 &request,
3931 &result,
3932 None::<String>,
3933 WorkflowTransactionClass::NonIdempotent,
3934 Some(VerificationPredicate::TextContains {
3935 value: "Order submitted".into(),
3936 }),
3937 )
3938 .unwrap();
3939
3940 let draft = recorder.draft();
3941 let step = &draft.steps[0];
3942 assert_eq!(
3943 step.target.as_ref().unwrap().region_kind,
3944 Some(SemanticRegionKind::CheckoutSummary)
3945 );
3946 assert_eq!(step.semantic.as_ref().unwrap().revision, Some(7));
3947 assert!(step.semantic.as_ref().unwrap().target_fingerprint.is_some());
3948 let serialized = serde_json::to_string(draft).unwrap();
3949 assert!(!serialized.contains("backend-secret"));
3950 assert!(!serialized.contains("token=secret"));
3951 assert!(!serialized.contains("target-7"));
3952 assert!(!serialized.contains("frame-2"));
3953
3954 let definition = recorder
3955 .into_definition(
3956 BTreeMap::new(),
3957 WorkflowBudgets {
3958 max_steps: 1,
3959 max_duration_ms: 30_000,
3960 max_retries: 0,
3961 max_extracted_bytes: 4_096,
3962 },
3963 VerificationPredicate::TextContains {
3964 value: "Order submitted".into(),
3965 },
3966 BTreeMap::new(),
3967 )
3968 .unwrap();
3969 assert!(definition.steps[0].intent.is_some());
3970 assert!(definition.steps[0].expect.is_some());
3971 }
3972
3973 #[test]
3974 fn recorder_keeps_ambiguous_semantic_results_unselected() {
3975 let request = SemanticIntentRequest {
3976 schema_version: INTENT_RESOLUTION_SCHEMA_VERSION,
3977 intent: "open settings".into(),
3978 action: SemanticIntentAction::Click,
3979 scope: IntentScope::default(),
3980 constraints: IntentConstraints::default(),
3981 resolution_policy: SemanticResolutionPolicy::RequireUniqueHighConfidence,
3982 expected_revision: None,
3983 };
3984 let candidate = |id: &str| SemanticIntentCandidate {
3985 id: id.into(),
3986 reference: format!("{id}:ref"),
3987 role: "button".into(),
3988 name: "Settings".into(),
3989 input_type: None,
3990 region_id: None,
3991 region_kind: None,
3992 confidence: IntentConfidence::High,
3993 evidence: Vec::new(),
3994 fingerprint: None,
3995 };
3996 let result = SemanticIntentResult {
3997 schema_version: INTENT_RESOLUTION_SCHEMA_VERSION,
3998 intent: request.intent.clone(),
3999 action: request.action,
4000 normalized_intent: request.intent.clone(),
4001 resolution: SemanticResolution::Ambiguous,
4002 policy_decision: IntentPolicyDecision::Rejected,
4003 route: None,
4004 revision: Some(3),
4005 candidates: vec![candidate("one"), candidate("two")],
4006 excluded_candidates: Vec::new(),
4007 excluded_count: 0,
4008 selected_candidate: None,
4009 suggested_constraints: Vec::new(),
4010 reason: Some("two candidates remain".into()),
4011 };
4012 let mut recorder = WorkflowRecorder::new("settings", "1.0.0");
4013 recorder
4014 .record_semantic_intent(
4015 "open-settings",
4016 &request,
4017 &result,
4018 None::<String>,
4019 WorkflowTransactionClass::Unknown,
4020 None,
4021 )
4022 .unwrap();
4023 let step = &recorder.draft().steps[0];
4024 assert!(step.target.is_none());
4025 assert_eq!(step.confidence, WorkflowRecordingConfidence::Low);
4026 assert!(step.semantic.as_ref().unwrap().ambiguous);
4027 assert!(step.review_required);
4028 }
4029
4030 #[test]
4031 fn workflow_step_flattens_batch_action() {
4032 let value = serde_json::to_value(&definition().steps[0]).unwrap();
4033 assert_eq!(value["id"], "open");
4034 assert_eq!(value["action"], "navigate");
4035 assert_eq!(value["timeoutMs"], 20_000);
4036 }
4037
4038 #[test]
4039 fn state_machine_accepts_linear_commit_and_rejects_invalid_jump() {
4040 assert!(WorkflowStepState::Pending.can_transition_to(WorkflowStepState::Ready));
4041 assert!(WorkflowStepState::Resolving.can_transition_to(WorkflowStepState::Dispatched));
4042 assert!(!WorkflowStepState::Pending.can_transition_to(WorkflowStepState::Committed));
4043
4044 let mut record = WorkflowStepRecord::new("open");
4045 for state in [
4046 WorkflowStepState::Ready,
4047 WorkflowStepState::Preflight,
4048 WorkflowStepState::Resolving,
4049 WorkflowStepState::Dispatched,
4050 WorkflowStepState::EffectObserved,
4051 WorkflowStepState::Verified,
4052 WorkflowStepState::OutputsExtracted,
4053 WorkflowStepState::Committed,
4054 ] {
4055 record.transition(state).unwrap();
4056 }
4057 assert_eq!(record.state, WorkflowStepState::Committed);
4058 assert_eq!(record.history.len(), 9);
4059 assert!(record.transition(WorkflowStepState::Preflight).is_err());
4060 let trace = WorkflowTrace::from_steps(&[record]);
4061 trace.validate().unwrap();
4062 assert_eq!(trace.events[0].sequence, 0);
4063 assert_eq!(
4064 trace.events.last().unwrap().state,
4065 WorkflowStepState::Committed
4066 );
4067 }
4068
4069 #[test]
4070 fn step_record_serializes_bounded_execution_evidence() {
4071 let mut record = WorkflowStepRecord::new("open");
4072 record.execution_ids = vec!["act_7".into(), "act_8".into()];
4073 record.dispatch_acknowledged = true;
4074 record.effect_observed = true;
4075 record.postcondition_verified = true;
4076 record.retry_safe = false;
4077 record.previous_revision = Some(7);
4078 record.current_revision = Some(8);
4079 record.intent_evidence = Some(WorkflowIntentEvidence {
4080 resolution_id: "res_1".into(),
4081 candidate_id: "candidate_1".into(),
4082 revision: 8,
4083 resolution: crate::browser::session::SemanticResolution::Exact,
4084 policy_decision: crate::browser::session::IntentPolicyDecision::Allowed,
4085 confidence: crate::browser::session::IntentConfidence::Exact,
4086 fingerprint: None,
4087 });
4088
4089 let trace = WorkflowTrace::from_steps(std::slice::from_ref(&record));
4090 assert_eq!(trace.intent_resolutions.len(), 1);
4091 let value = serde_json::to_value(record).unwrap();
4092 assert_eq!(value["executionIds"], serde_json::json!(["act_7", "act_8"]));
4093 assert_eq!(value["dispatchAcknowledged"], true);
4094 assert_eq!(value["effectObserved"], true);
4095 assert_eq!(value["postconditionVerified"], true);
4096 assert_eq!(value["retrySafe"], false);
4097 assert_eq!(value["previousRevision"], 7);
4098 assert_eq!(value["currentRevision"], 8);
4099 assert_eq!(value["intentEvidence"]["candidateId"], "candidate_1");
4100 }
4101
4102 #[test]
4103 fn budget_exhaustion_is_a_typed_run_status() {
4104 let result = WorkflowRunResult::budget_exhausted(
4105 &definition(),
4106 "run_test".into(),
4107 vec![WorkflowStepRecord::new("open")],
4108 Some("open".into()),
4109 "maxSteps exhausted",
4110 3,
4111 3,
4112 );
4113 assert_eq!(result.status, WorkflowRunStatus::BudgetExhausted);
4114 assert_eq!(result.trace.run_id.as_deref(), Some("run_test"));
4115 let value = serde_json::to_value(result).unwrap();
4116 assert_eq!(value["status"], "budget_exhausted");
4117 }
4118
4119 #[test]
4120 fn trace_schema_version_is_independent_and_validated() {
4121 let trace = WorkflowTrace::from_steps(&[]);
4122 assert_eq!(trace.schema_version, WORKFLOW_TRACE_SCHEMA_VERSION);
4123 trace.validate().unwrap();
4124
4125 let mut unsupported = trace;
4126 unsupported.schema_version = WORKFLOW_TRACE_SCHEMA_VERSION + 1;
4127 let error = unsupported.validate().unwrap_err();
4128 assert_eq!(error.path, "trace.schemaVersion");
4129 }
4130
4131 #[test]
4132 fn post_dispatch_failures_require_resume_reconciliation() {
4133 let result = WorkflowRunResult::resume_required(
4134 &definition(),
4135 "run_test".into(),
4136 vec![WorkflowStepRecord::new("open")],
4137 Some("open".into()),
4138 "postcondition was not proven",
4139 3,
4140 4,
4141 );
4142 assert_eq!(result.status, WorkflowRunStatus::ResumeRequired);
4143 assert_eq!(
4144 serde_json::to_value(result).unwrap()["status"],
4145 "resume_required"
4146 );
4147 }
4148
4149 #[test]
4150 fn trace_replay_preserves_attempt_boundaries() {
4151 let mut record = WorkflowStepRecord::new("open");
4152 record.transition(WorkflowStepState::Ready).unwrap();
4153 record.transition(WorkflowStepState::Preflight).unwrap();
4154 record.attempts = 1;
4155 record.transition(WorkflowStepState::Resolving).unwrap();
4156 record.transition(WorkflowStepState::NotDispatched).unwrap();
4157 record.fail(WorkflowStepState::FailedBeforeDispatch, "before dispatch");
4158 record.transition(WorkflowStepState::Ready).unwrap();
4159 record.transition(WorkflowStepState::Preflight).unwrap();
4160 record.attempts = 2;
4161 record.transition(WorkflowStepState::Resolving).unwrap();
4162 record.transition(WorkflowStepState::Dispatched).unwrap();
4163 record
4164 .transition(WorkflowStepState::EffectObserved)
4165 .unwrap();
4166 record.transition(WorkflowStepState::Verified).unwrap();
4167 record
4168 .transition(WorkflowStepState::OutputsExtracted)
4169 .unwrap();
4170 record.transition(WorkflowStepState::Committed).unwrap();
4171
4172 let workflow = definition();
4173 let trace = WorkflowTrace::from_steps(&[record]);
4174 let replayed = trace.replay(&workflow).unwrap();
4175 assert_eq!(trace.events[2].attempt, 1);
4176 assert_eq!(trace.events[7].attempt, 2);
4177 assert_eq!(replayed[0].state, WorkflowStepState::Committed);
4178 assert_eq!(replayed[0].attempts, 2);
4179 }
4180
4181 #[test]
4182 fn trace_replay_rejects_a_prefix_that_skips_the_first_step() {
4183 let mut workflow = definition();
4184 workflow.steps.push(WorkflowStep {
4185 id: "save".into(),
4186 action: BatchStep::Screenshot,
4187 intent: None,
4188 when: None,
4189 expect: None,
4190 before_retry: None,
4191 transaction: WorkflowTransactionClass::ReadOnly,
4192 idempotency_key: None,
4193 max_retries: 0,
4194 repeat: 1,
4195 });
4196 let trace = WorkflowTrace::from_steps(&[WorkflowStepRecord::new("save")]);
4197 let error = trace.replay(&workflow).unwrap_err();
4198 assert_eq!(error.path, "trace.events[0].stepId");
4199 }
4200
4201 #[test]
4202 fn conditional_steps_record_and_replay_branch_decisions() {
4203 let mut workflow = definition();
4204 let predicate = VerificationPredicate::TitleContains {
4205 value: "Example".into(),
4206 };
4207 workflow.steps[0].when = Some(predicate.clone());
4208 let mut record = WorkflowStepRecord::new("open");
4209 record.transition(WorkflowStepState::Ready).unwrap();
4210 record.branch_decision = Some(WorkflowBranchDecision {
4211 step_id: "open".into(),
4212 predicate: predicate.clone(),
4213 matched: false,
4214 });
4215 record.transition(WorkflowStepState::Skipped).unwrap();
4216 let trace = WorkflowTrace::from_steps(&[record]);
4217 assert!(!trace.branch_decisions[0].matched);
4218 let replayed = trace.replay(&workflow).unwrap();
4219 assert_eq!(replayed[0].state, WorkflowStepState::Skipped);
4220 assert_eq!(
4221 replayed[0].branch_decision.as_ref().unwrap().predicate,
4222 predicate
4223 );
4224 }
4225}