1use std::collections::{BTreeMap, BTreeSet};
14
15use serde_json::Value;
16
17pub const BUILTIN_TYPES: [&str; 4] = ["project", "item", "signal", "run"];
21
22pub const SYSTEM_FIELDS: [&str; 10] = [
25 "id", "type", "node", "parent", "root", "depth", "lap", "children", "project", "run",
26];
27
28pub const CHECK_RULES: [&str; 6] = [
31 "verify",
32 "scope-clean",
33 "lint-todo",
34 "ci-green",
35 "pr-merged",
36 "nonempty",
37];
38
39const MAP_RULES: [&str; 4] = ["todo-items", "open-issues", "open-runs", "outdated-deps"];
42const MAP_RULE_PREFIXES: [&str; 4] = ["as:", "where:", "path-present:", "path-absent:"];
43
44const REDUCE_RULES: [&str; 2] = ["dedupe", "rank"];
46const REDUCE_RULE_PREFIXES: [&str; 1] = ["limit:"];
47
48#[derive(Debug, Clone, Copy, PartialEq, Eq)]
50pub enum Out {
51 Grows,
53 Same,
55 Shrinks,
57}
58
59impl Out {
60 fn parse(s: &str) -> Option<Out> {
61 match s {
62 "0..n" => Some(Out::Grows),
63 "1" => Some(Out::Same),
64 "0..1" => Some(Out::Shrinks),
65 _ => None,
66 }
67 }
68
69 pub fn name(self) -> &'static str {
70 match self {
71 Out::Grows => "0..n",
72 Out::Same => "1",
73 Out::Shrinks => "0..1",
74 }
75 }
76}
77
78#[derive(Debug, Clone, Copy, PartialEq, Eq)]
81pub enum Via {
82 Agent,
83 Rule,
84}
85
86#[derive(Debug, Clone, PartialEq, Eq)]
90pub enum Count {
91 Fixed(i64),
92 Param(String),
93}
94
95impl Count {
96 pub fn ceiling(&self, params: &BTreeMap<String, Param>) -> i64 {
98 match self {
99 Count::Fixed(n) => *n,
100 Count::Param(p) => params.get(p).and_then(|p| p.max).unwrap_or(0),
101 }
102 }
103}
104
105#[derive(Debug, Clone, PartialEq)]
106pub enum FieldType {
107 Line { max: i64, unique: bool },
108 Lines { max: i64 },
109 Enum { values: Vec<String> },
110 List,
111 Int { min: Option<i64>, max: Option<i64> },
112 Bool,
113}
114
115#[derive(Debug, Clone, PartialEq)]
116pub struct Field {
117 pub ty: FieldType,
118 pub required: bool,
119}
120
121const LIST_MAX: usize = 20;
123const LIST_ENTRY_MAX: usize = 200;
124
125impl FieldType {
126 pub fn describe(&self) -> String {
128 match self {
129 FieldType::Line { max, .. } => format!("one line of text, at most {max} characters"),
130 FieldType::Lines { max } => {
131 format!("a JSON array of at most {max} strings, one line each")
132 }
133 FieldType::Enum { values } => format!("one of {}", values.join(", ")),
134 FieldType::List => format!(
135 "a JSON array of at most {LIST_MAX} strings, each at most {LIST_ENTRY_MAX} characters"
136 ),
137 FieldType::Int { min, max } => match (min, max) {
138 (Some(lo), Some(hi)) => format!("a whole number from {lo} to {hi}"),
139 (Some(lo), None) => format!("a whole number, at least {lo}"),
140 (None, Some(hi)) => format!("a whole number, at most {hi}"),
141 (None, None) => "a whole number".into(),
142 },
143 FieldType::Bool => "true or false".into(),
144 }
145 }
146
147 fn check(&self, name: &str, v: &Value) -> Result<(), String> {
148 let at = |e: String| format!("`{name}`: {e}");
149 match self {
150 FieldType::Line { max, .. } => {
151 let s = v.as_str().ok_or_else(|| at("expected a string".into()))?;
152 if s.contains('\n') {
153 return Err(at("expected one line".into()));
154 }
155 if s.is_empty() || s.chars().count() > *max as usize {
156 return Err(at(format!("expected 1 to {max} characters")));
157 }
158 Ok(())
159 }
160 FieldType::Lines { max } => {
161 let lines = v.as_array().ok_or_else(|| at("expected an array".into()))?;
162 if lines.len() > *max as usize {
163 return Err(at(format!("expected at most {max} lines")));
164 }
165 match lines.iter().all(Value::is_string) {
166 true => Ok(()),
167 false => Err(at("every line must be a string".into())),
168 }
169 }
170 FieldType::Enum { values } => {
171 let s = v.as_str().ok_or_else(|| at("expected a string".into()))?;
172 match values.iter().any(|x| x == s) {
173 true => Ok(()),
174 false => Err(at(format!("`{s}` is not one of {}", values.join(", ")))),
175 }
176 }
177 FieldType::List => {
178 let list = v.as_array().ok_or_else(|| at("expected an array".into()))?;
179 if list.len() > LIST_MAX {
180 return Err(at(format!("expected at most {LIST_MAX} entries")));
181 }
182 match list.iter().all(|e| {
183 e.as_str()
184 .is_some_and(|s| s.chars().count() <= LIST_ENTRY_MAX)
185 }) {
186 true => Ok(()),
187 false => Err(at(format!(
188 "every entry must be a string of at most {LIST_ENTRY_MAX} characters"
189 ))),
190 }
191 }
192 FieldType::Int { min, max } => {
193 let n = v
194 .as_i64()
195 .ok_or_else(|| at("expected a whole number".into()))?;
196 if min.is_some_and(|m| n < m) || max.is_some_and(|m| n > m) {
197 return Err(at(format!(
198 "expected {} to {}",
199 min.map_or("any".into(), |m| m.to_string()),
200 max.map_or("any".into(), |m| m.to_string())
201 )));
202 }
203 Ok(())
204 }
205 FieldType::Bool => match v.is_boolean() {
206 true => Ok(()),
207 false => Err(at("expected true or false".into())),
208 },
209 }
210 }
211}
212
213#[derive(Debug, Clone, PartialEq, Default)]
214pub struct Type {
215 pub fields: BTreeMap<String, Field>,
216}
217
218impl Type {
219 pub fn check(&self, v: &Value) -> Result<(), String> {
223 let Some(map) = v.as_object() else {
224 return Err("a unit must be an object".into());
225 };
226 if let Some(unknown) = map
229 .keys()
230 .find(|k| !k.starts_with('@') && !self.fields.contains_key(*k))
231 {
232 return Err(format!("unknown field `{unknown}`"));
233 }
234 for (name, field) in &self.fields {
235 let Some(value) = map.get(name).filter(|v| !v.is_null()) else {
236 if field.required {
237 return Err(format!("`{name}` is required"));
238 }
239 continue;
240 };
241 field.ty.check(name, value)?;
242 }
243 Ok(())
244 }
245
246 fn of(names: &[&str]) -> Type {
247 Type {
248 fields: names
249 .iter()
250 .map(|n| {
251 (
252 (*n).to_string(),
253 Field {
254 ty: FieldType::Line {
255 max: 4096,
256 unique: false,
257 },
258 required: false,
259 },
260 )
261 })
262 .collect(),
263 }
264 }
265}
266
267pub fn builtin_type(name: &str) -> Option<Type> {
270 let fields: &[&str] = match name {
271 "project" => &[
272 "name",
273 "tier",
274 "repo",
275 "owner",
276 "default_branch",
277 "ci",
278 "deps",
279 "tags",
280 ],
281 "item" => &[
282 "key",
283 "text",
284 "priority",
285 "tags",
286 "due",
287 "gh",
288 "group",
289 "description",
290 "done",
291 ],
292 "signal" => &["kind", "detail"],
293 "run" => &["id", "task", "state", "branch", "pr", "verify", "cost"],
294 _ => return None,
295 };
296 Some(Type::of(fields))
297}
298
299#[derive(Debug, Clone, PartialEq)]
300pub enum ParamType {
301 Name,
303 Line,
304 Int,
305 Enum(Vec<String>),
306 Bool,
307 List,
308}
309
310impl ParamType {
311 pub fn name(&self) -> &'static str {
312 match self {
313 ParamType::Name => "name",
314 ParamType::Line => "line",
315 ParamType::Int => "int",
316 ParamType::Bool => "bool",
317 ParamType::List => "list",
318 ParamType::Enum(_) => "enum",
319 }
320 }
321
322 pub fn read(&self, text: &str) -> Result<Value, String> {
325 match self {
326 ParamType::Name => {
327 let ok = !text.is_empty()
328 && text != "."
329 && text != ".."
330 && text
331 .chars()
332 .all(|c| c.is_ascii_alphanumeric() || "._-".contains(c));
333 match ok {
334 true => Ok(Value::from(text)),
335 false => Err("expected a filename, such as `REVIEW.md`".into()),
336 }
337 }
338 ParamType::Line => match !text.is_empty() && !text.contains('\n') {
339 true => Ok(Value::from(text)),
340 false => Err("expected one non-empty line".into()),
341 },
342 ParamType::Int => text
343 .parse::<i64>()
344 .map(Value::from)
345 .map_err(|_| "expected a whole number".to_string()),
346 ParamType::Bool => match text {
347 "true" => Ok(Value::Bool(true)),
348 "false" => Ok(Value::Bool(false)),
349 _ => Err("expected true or false".into()),
350 },
351 ParamType::List => {
352 let parts: Vec<&str> = match text.is_empty() {
353 true => Vec::new(),
354 false => text.split(',').map(str::trim).collect(),
355 };
356 if parts.len() > LIST_MAX {
357 return Err(format!("expected at most {LIST_MAX} entries"));
358 }
359 match parts.iter().all(|p| p.chars().count() <= LIST_ENTRY_MAX) {
360 true => Ok(Value::from(parts)),
361 false => Err(format!(
362 "every entry must be at most {LIST_ENTRY_MAX} characters"
363 )),
364 }
365 }
366 ParamType::Enum(values) => match values.iter().any(|v| v == text) {
367 true => Ok(Value::from(text)),
368 false => Err(format!("expected one of {}", values.join(", "))),
369 },
370 }
371 }
372
373 pub fn holds(&self, v: &Value) -> Result<(), String> {
376 match (self, v) {
377 (ParamType::Int, Value::Number(n)) if n.is_i64() => Ok(()),
378 (ParamType::Bool, Value::Bool(_)) => Ok(()),
379 (ParamType::List, Value::Array(items)) => {
380 match items.len() <= LIST_MAX && items.iter().all(Value::is_string) {
381 true => Ok(()),
382 false => Err(format!("expected at most {LIST_MAX} strings")),
383 }
384 }
385 (ParamType::Name | ParamType::Line | ParamType::Enum(_), Value::String(s)) => {
386 self.read(s).map(|_| ())
387 }
388 _ => Err(format!("expected {}", self.name())),
389 }
390 }
391}
392
393#[derive(Debug, Clone, PartialEq)]
394pub struct Param {
395 pub ty: ParamType,
396 pub default: Value,
397 pub max: Option<i64>,
400}
401
402#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
403pub struct Effects {
404 pub repo: bool,
406 pub writes: bool,
408}
409
410impl Effects {
411 fn union(self, other: Effects) -> Effects {
412 Effects {
413 repo: self.repo || other.repo,
414 writes: self.writes || other.writes,
415 }
416 }
417
418 pub fn names(self) -> Vec<&'static str> {
419 let mut v = Vec::new();
420 if self.repo {
421 v.push("repo");
422 }
423 if self.writes {
424 v.push("writes");
425 }
426 v
427 }
428}
429
430#[derive(Debug, Clone, Copy, PartialEq, Eq)]
431pub struct Caps {
432 pub max_units: i64,
433 pub max_edits: i64,
434}
435
436#[derive(Debug, Clone, PartialEq)]
437pub struct Retry {
438 pub max: Count,
439 pub predicate: String,
440 pub escalate: Option<String>,
441}
442
443#[derive(Debug, Clone, Copy, PartialEq, Eq)]
444pub enum Sink {
445 Todo,
446 Note,
447 Doc,
448}
449
450#[derive(Debug, Clone, Copy, PartialEq, Eq)]
451pub enum Action {
452 Add,
453 Tick,
454 Remove,
455}
456
457#[derive(Debug, Clone, PartialEq)]
458pub struct MapNode {
459 pub via: Via,
460 pub out: Out,
461 pub emits: String,
462 pub writes: Vec<String>,
463 pub max_units: Option<Count>,
464 pub max_depth: Option<Count>,
465 pub task: Option<String>,
466 pub rule: Option<String>,
467 pub doc: Option<String>,
468 pub publish: bool,
469 pub retry: Option<Retry>,
470}
471
472#[derive(Debug, Clone, PartialEq)]
473pub struct ReduceNode {
474 pub via: Via,
475 pub emits: String,
476 pub group_by: Vec<String>,
477 pub task: Option<String>,
478 pub rule: Option<String>,
479}
480
481#[derive(Debug, Clone, PartialEq)]
482pub struct EditNode {
483 pub task: String,
484 pub check: Option<String>,
485 pub retry: Option<Retry>,
486}
487
488#[derive(Debug, Clone, PartialEq)]
489pub struct EmitNode {
490 pub sink: Sink,
491 pub action: Action,
492 pub map: BTreeMap<String, String>,
493}
494
495#[derive(Debug, Clone, PartialEq)]
496pub struct CallNode {
497 pub workflow: String,
498 pub with: BTreeMap<String, Value>,
499}
500
501#[derive(Debug, Clone, PartialEq)]
502pub enum Op {
503 Map(MapNode),
504 Reduce(ReduceNode),
505 Edit(EditNode),
506 Check { rule: String },
507 Emit(EmitNode),
508 Call(CallNode),
509}
510
511impl Op {
512 pub fn name(&self) -> &'static str {
513 match self {
514 Op::Map(_) => "map",
515 Op::Reduce(_) => "reduce",
516 Op::Edit(_) => "edit",
517 Op::Check { .. } => "check",
518 Op::Emit(_) => "emit",
519 Op::Call(_) => "call",
520 }
521 }
522}
523
524#[derive(Debug, Clone, PartialEq)]
525pub struct Node {
526 pub name: String,
527 pub input: String,
529 pub op: Op,
530}
531
532#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
533pub enum From {
534 Input,
535 Node(String),
536}
537
538#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
539pub enum To {
540 Output,
541 Node(String),
542}
543
544#[derive(Debug, Clone, Copy, PartialEq, Eq)]
545pub enum Cmp {
546 Eq,
547 Lt,
548 Le,
549 Gt,
550 Ge,
551}
552
553impl Cmp {
554 fn parse(s: &str) -> Option<Cmp> {
555 match s {
556 "==" => Some(Cmp::Eq),
557 "<" => Some(Cmp::Lt),
558 "<=" => Some(Cmp::Le),
559 ">" => Some(Cmp::Gt),
560 ">=" => Some(Cmp::Ge),
561 _ => None,
562 }
563 }
564}
565
566#[derive(Debug, Clone, PartialEq, Eq)]
569pub enum Key {
570 Field(String),
571 System(String),
572}
573
574#[derive(Debug, Clone, PartialEq)]
575pub enum Test {
576 In(Vec<String>),
578 Compare(Cmp, Count),
579 Present,
580 Absent,
581}
582
583#[derive(Debug, Clone, PartialEq, Default)]
584pub struct Guard(pub Vec<(Key, Test)>);
585
586#[derive(Debug, Clone, PartialEq)]
587pub struct Edge {
588 pub from: From,
589 pub to: To,
590 pub when: Option<Guard>,
591 pub default: bool,
593 pub max_laps: Option<Count>,
595}
596
597impl Edge {
598 pub fn is_forward(&self) -> bool {
600 self.max_laps.is_none()
601 && !matches!((&self.from, &self.to), (From::Node(a), To::Node(b)) if a == b)
602 }
603}
604
605#[derive(Debug, Clone, PartialEq)]
606pub struct Workflow {
607 pub name: String,
608 pub input: String,
609 pub output: Option<String>,
610 pub effects: Effects,
611 pub params: BTreeMap<String, Param>,
612 pub caps: Caps,
613 pub nodes: Vec<Node>,
614 pub edges: Vec<Edge>,
615}
616
617impl Workflow {
618 pub fn node(&self, name: &str) -> Option<&Node> {
619 self.nodes.iter().find(|n| n.name == name)
620 }
621
622 pub fn output_of(&self, node: &Node) -> String {
624 match &node.op {
625 Op::Map(m) => m.emits.clone(),
626 Op::Reduce(r) => r.emits.clone(),
627 _ => node.input.clone(),
628 }
629 }
630
631 pub fn predecessors(&self, node: &str) -> Vec<&str> {
633 self.edges
634 .iter()
635 .filter(|e| e.is_forward() && e.to == To::Node(node.to_string()))
636 .filter_map(|e| match &e.from {
637 From::Node(n) => Some(n.as_str()),
638 From::Input => None,
639 })
640 .collect()
641 }
642}
643
644#[derive(Debug, Clone, PartialEq, Default)]
645pub struct Document {
646 pub types: BTreeMap<String, Type>,
647 pub workflows: Vec<Workflow>,
648}
649
650impl Document {
651 pub fn workflow(&self, name: &str) -> Option<&Workflow> {
652 self.workflows.iter().find(|w| w.name == name)
653 }
654
655 pub fn resolve(&self, name: &str) -> Option<Type> {
657 builtin_type(name).or_else(|| self.types.get(name).cloned())
658 }
659
660 pub fn parse(text: &str) -> Result<Document, String> {
664 let v: Value = serde_json::from_str(text).map_err(|e| format!("not JSON: {e}"))?;
665 known(&v, &["types", "workflow"], "document")?;
666 let mut types = BTreeMap::new();
667 if let Some(map) = v.get("types") {
668 let map = map
669 .as_object()
670 .ok_or_else(|| "`types` must be an object".to_string())?;
671 for (name, decl) in map {
672 types.insert(name.clone(), parse_type(name, decl)?);
673 }
674 }
675 let list = v
676 .get("workflow")
677 .and_then(Value::as_array)
678 .ok_or("expected a `workflow` array")?;
679 if list.is_empty() {
680 return Err("a document with no workflow defines nothing".into());
681 }
682 let workflows = list
683 .iter()
684 .map(parse_workflow)
685 .collect::<Result<Vec<_>, String>>()?;
686 let doc = Document { types, workflows };
687 doc.validate()?;
688 Ok(doc)
689 }
690}
691
692fn known(v: &Value, fields: &[&str], what: &str) -> Result<(), String> {
693 let Some(map) = v.as_object() else {
694 return Err(format!("{what} must be an object"));
695 };
696 match map.keys().find(|k| !fields.contains(&k.as_str())) {
697 Some(k) => Err(format!(
698 "{what}: unknown field `{k}`; expected one of {}",
699 fields.join(", ")
700 )),
701 None => Ok(()),
702 }
703}
704
705fn parse_type(name: &str, v: &Value) -> Result<Type, String> {
706 let at = |e: String| format!("type `{name}`: {e}");
707 if BUILTIN_TYPES.contains(&name) {
708 return Err(at("that is a built-in type; choose another name".into()));
709 }
710 known(v, &["fields"], &format!("type `{name}`"))?;
711 let fields = v
712 .get("fields")
713 .and_then(Value::as_object)
714 .ok_or_else(|| at("no `fields`".into()))?;
715 let mut out = BTreeMap::new();
716 for (field, decl) in fields {
717 if field.starts_with('@') {
718 return Err(at(format!("field `{field}` is reserved")));
719 }
720 known(
721 decl,
722 &["type", "required", "max", "min", "values", "unique"],
723 &format!("type `{name}` field `{field}`"),
724 )?;
725 let kind = decl["type"]
726 .as_str()
727 .ok_or_else(|| at(format!("field `{field}` has no `type`")))?;
728 let whole = |key: &str| -> Result<Option<i64>, String> {
731 match &decl[key] {
732 Value::Null => Ok(None),
733 v => v
734 .as_i64()
735 .map(Some)
736 .ok_or_else(|| at(format!("field `{field}`: `{key}` must be a whole number"))),
737 }
738 };
739 let max = whole("max")?;
740 let min = whole("min")?;
741 if !matches!(kind, "int") && min.is_some() {
742 return Err(at(format!(
743 "field `{field}`: `min` belongs to an int, and this is {kind}"
744 )));
745 }
746 if matches!(kind, "enum" | "list" | "bool") && max.is_some() {
747 return Err(at(format!(
748 "field `{field}`: `max` does not apply to {kind}"
749 )));
750 }
751 if matches!(kind, "line" | "lines") && max.is_some_and(|m| m < 1) {
752 return Err(at(format!("field `{field}`: `max` must be 1 or more")));
753 }
754 let unique = match &decl["unique"] {
755 Value::Null => false,
756 Value::String(s) if s == "normalised" && kind == "line" => true,
757 _ => {
758 return Err(at(format!(
759 "field `{field}`: `unique` takes `normalised`, on a line"
760 )));
761 }
762 };
763 if kind != "enum" && !decl["values"].is_null() {
764 return Err(at(format!(
765 "field `{field}`: `values` belongs to an enum, and this is {kind}"
766 )));
767 }
768 if !decl["required"].is_null() && !decl["required"].is_boolean() {
769 return Err(at(format!(
770 "field `{field}`: `required` must be true or false"
771 )));
772 }
773 let ty = match kind {
774 "line" => FieldType::Line {
775 max: max.unwrap_or(200),
776 unique,
777 },
778 "lines" => FieldType::Lines {
779 max: max.unwrap_or(40),
780 },
781 "enum" => {
782 let values = decl["values"]
783 .as_array()
784 .ok_or_else(|| at(format!("field `{field}` is an enum with no `values`")))?
785 .iter()
786 .map(|v| {
787 v.as_str()
788 .map(String::from)
789 .ok_or_else(|| at(format!("field `{field}`: values must be strings")))
790 })
791 .collect::<Result<Vec<_>, String>>()?;
792 if values.is_empty() {
793 return Err(at(format!("field `{field}`: an enum with no values")));
794 }
795 FieldType::Enum { values }
796 }
797 "list" => FieldType::List,
798 "int" => {
799 if let (Some(lo), Some(hi)) = (min, max)
800 && lo > hi
801 {
802 return Err(at(format!("field `{field}`: `min` is over `max`")));
803 }
804 FieldType::Int { min, max }
805 }
806 "bool" => FieldType::Bool,
807 other => return Err(at(format!("field `{field}`: unknown type `{other}`"))),
808 };
809 out.insert(
810 field.clone(),
811 Field {
812 ty,
813 required: decl["required"].as_bool().unwrap_or(false),
814 },
815 );
816 }
817 Ok(Type { fields: out })
818}
819
820fn count(v: &Value, what: &str) -> Result<Count, String> {
821 match v {
822 Value::Number(n) => n
823 .as_i64()
824 .map(Count::Fixed)
825 .ok_or_else(|| format!("{what} must be a whole number")),
826 Value::String(s) => match param_ref(s) {
827 Some(p) => Ok(Count::Param(p)),
828 None => Err(format!(
829 "{what} must be a number or `{{$param}}`, not `{s}`"
830 )),
831 },
832 _ => Err(format!("{what} must be a number or `{{$param}}`")),
833 }
834}
835
836fn param_ref(s: &str) -> Option<String> {
839 let inner = s.strip_prefix("{$")?.strip_suffix('}')?;
840 (!inner.is_empty() && !inner.contains(['{', '}'])).then(|| inner.to_string())
841}
842
843fn parse_params(
844 v: &Value,
845 at: &dyn Fn(String) -> String,
846) -> Result<BTreeMap<String, Param>, String> {
847 let mut out = BTreeMap::new();
848 let Some(map) = v.as_object() else {
849 return Err(at("`params` must be an object".into()));
850 };
851 for (name, decl) in map {
852 if name.contains('/') {
854 return Err(at(format!(
855 "parameter `{name}`: a name may not contain `/`"
856 )));
857 }
858 known(
859 decl,
860 &["type", "default", "values", "max", "min"],
861 &format!("parameter `{name}`"),
862 )
863 .map_err(at)?;
864 let kind = decl["type"]
865 .as_str()
866 .ok_or_else(|| at(format!("parameter `{name}` has no `type`")))?;
867 let ty = match kind {
868 "name" => ParamType::Name,
869 "line" => ParamType::Line,
870 "int" => ParamType::Int,
871 "bool" => ParamType::Bool,
872 "list" => ParamType::List,
873 "enum" => ParamType::Enum(
874 decl["values"]
875 .as_array()
876 .ok_or_else(|| at(format!("parameter `{name}` is an enum with no `values`")))?
877 .iter()
878 .map(|v| {
879 v.as_str().map(String::from).ok_or_else(|| {
880 at(format!("parameter `{name}`: values must be strings"))
881 })
882 })
883 .collect::<Result<Vec<_>, String>>()?,
884 ),
885 other => return Err(at(format!("parameter `{name}`: unknown type `{other}`"))),
886 };
887 if let ParamType::Enum(values) = &ty
888 && values.is_empty()
889 {
890 return Err(at(format!("parameter `{name}`: an enum with no values")));
891 }
892 if !matches!(ty, ParamType::Enum(_)) && !decl["values"].is_null() {
893 return Err(at(format!(
894 "parameter `{name}`: `values` belongs to an enum, and this is {kind}"
895 )));
896 }
897 let max = match &decl["max"] {
900 Value::Null => None,
901 _ if !matches!(ty, ParamType::Int) => {
902 return Err(at(format!(
903 "parameter `{name}`: `max` bounds an int, and this is {kind}"
904 )));
905 }
906 v => Some(
907 v.as_i64()
908 .filter(|n| *n >= 0)
909 .ok_or_else(|| at(format!("parameter `{name}`: `max` must be 0 or more")))?,
910 ),
911 };
912 let default = decl.get("default").cloned().ok_or_else(|| {
913 at(format!(
914 "parameter `{name}` has no `default`; a workflow must be runnable with no arguments"
915 ))
916 })?;
917 ty.holds(&default)
920 .map_err(|e| at(format!("parameter `{name}`: default `{default}`: {e}")))?;
921 if let Some(max) = max
922 && default.as_i64().is_some_and(|d| d > max)
923 {
924 return Err(at(format!(
925 "parameter `{name}`: default {default} is over its own maximum of {max}"
926 )));
927 }
928 out.insert(name.clone(), Param { ty, default, max });
929 }
930 Ok(out)
931}
932
933fn parse_retry(v: &Value, at: &dyn Fn(String) -> String) -> Result<Retry, String> {
934 known(v, &["max", "while", "escalate"], "retry").map_err(at)?;
935 let max = count(&v["max"], "retry max").map_err(at)?;
936 let predicate = v["while"]
937 .as_str()
938 .ok_or_else(|| at("retry needs a `while`".into()))?
939 .to_string();
940 let escalate = match &v["escalate"] {
941 Value::Null => None,
942 e => Some(
943 e["model"]
944 .as_str()
945 .ok_or_else(|| at("escalate needs a model".into()))?
946 .to_string(),
947 ),
948 };
949 Ok(Retry {
950 max,
951 predicate,
952 escalate,
953 })
954}
955
956fn parse_node(wf: &str, v: &Value) -> Result<Node, String> {
957 let name = v["name"]
958 .as_str()
959 .ok_or_else(|| format!("workflow `{wf}`: a node with no `name`"))?
960 .to_string();
961 let at = |e: String| format!("workflow `{wf}`: node `{name}`: {e}");
962 known(
963 v,
964 &[
965 "name",
966 "op",
967 "via",
968 "in",
969 "out",
970 "emits",
971 "writes",
972 "max_units",
973 "max_depth",
974 "group_by",
975 "task",
976 "rule",
977 "doc",
978 "publish",
979 "check",
980 "retry",
981 "sink",
982 "action",
983 "map",
984 "workflow",
985 "with",
986 ],
987 &format!("workflow `{wf}` node `{name}`"),
988 )?;
989 let input = v["in"]
990 .as_str()
991 .ok_or_else(|| at("no `in` type".into()))?
992 .to_string();
993 let via = match v["via"].as_str() {
994 None => None,
995 Some("agent") => Some(Via::Agent),
996 Some("rule") => Some(Via::Rule),
997 Some(other) => return Err(at(format!("`via` is agent or rule, not `{other}`"))),
998 };
999 let task = v["task"].as_str().map(String::from);
1000 let rule = v["rule"].as_str().map(String::from);
1001 let retry = match &v["retry"] {
1002 Value::Null => None,
1003 r => Some(parse_retry(r, &at)?),
1004 };
1005 let strings = |key: &str| -> Result<Vec<String>, String> {
1006 match &v[key] {
1007 Value::Null => Ok(Vec::new()),
1008 Value::Array(a) => a
1009 .iter()
1010 .map(|s| {
1011 s.as_str()
1012 .map(String::from)
1013 .ok_or_else(|| at(format!("`{key}` takes field names")))
1014 })
1015 .collect(),
1016 _ => Err(at(format!("`{key}` must be a list"))),
1017 }
1018 };
1019 let op = match v["op"].as_str() {
1020 None => return Err(at("no `op`".into())),
1021 Some("map") => {
1022 let out = v["out"]
1023 .as_str()
1024 .and_then(Out::parse)
1025 .ok_or_else(|| at("`out` is 0..n, 1 or 0..1".into()))?;
1026 Op::Map(MapNode {
1027 via: via.unwrap_or(Via::Agent),
1028 out,
1029 emits: v["emits"]
1030 .as_str()
1031 .ok_or_else(|| at("a map states what it `emits`".into()))?
1032 .to_string(),
1033 writes: strings("writes")?,
1034 max_units: match &v["max_units"] {
1035 Value::Null => None,
1036 c => Some(count(c, "max_units").map_err(at)?),
1037 },
1038 max_depth: match &v["max_depth"] {
1039 Value::Null => None,
1040 c => Some(count(c, "max_depth").map_err(at)?),
1041 },
1042 task,
1043 rule,
1044 doc: v["doc"].as_str().map(String::from),
1045 publish: v["publish"].as_bool().unwrap_or(false),
1046 retry,
1047 })
1048 }
1049 Some("reduce") => Op::Reduce(ReduceNode {
1050 via: via.unwrap_or(Via::Agent),
1051 emits: v["emits"]
1052 .as_str()
1053 .ok_or_else(|| at("a reduce states what it `emits`".into()))?
1054 .to_string(),
1055 group_by: strings("group_by")?,
1056 task,
1057 rule,
1058 }),
1059 Some("edit") => Op::Edit(EditNode {
1060 task: task.ok_or_else(|| at("an edit needs a `task`".into()))?,
1061 check: v["check"].as_str().map(String::from),
1062 retry,
1063 }),
1064 Some("check") => Op::Check {
1065 rule: rule.ok_or_else(|| at("a check needs a `rule`".into()))?,
1066 },
1067 Some("emit") => {
1068 let sink = match v["sink"].as_str() {
1069 Some("todo") => Sink::Todo,
1070 Some("note") => Sink::Note,
1071 Some("doc") => Sink::Doc,
1072 Some("issue") => {
1073 return Err(at(
1074 "the issue sink is refused: `pma sync` owns issue creation".into(),
1075 ));
1076 }
1077 Some(other) => return Err(at(format!("unknown sink `{other}`"))),
1078 None => return Err(at("an emit names a `sink`".into())),
1079 };
1080 let action = match v["action"].as_str() {
1081 None | Some("add") => Action::Add,
1082 Some("tick") => Action::Tick,
1083 Some("remove") => Action::Remove,
1084 Some(other) => return Err(at(format!("unknown action `{other}`"))),
1085 };
1086 let map = v["map"]
1087 .as_object()
1088 .ok_or_else(|| at("an emit needs a `map` from sink field to placeholder".into()))?
1089 .iter()
1090 .map(|(k, val)| {
1091 val.as_str()
1092 .map(|s| (k.clone(), s.to_string()))
1093 .ok_or_else(|| at(format!("`map.{k}` must be a placeholder string")))
1094 })
1095 .collect::<Result<BTreeMap<_, _>, String>>()?;
1096 Op::Emit(EmitNode { sink, action, map })
1097 }
1098 Some("call") => Op::Call(CallNode {
1099 workflow: v["workflow"]
1100 .as_str()
1101 .ok_or_else(|| at("a call names a `workflow`".into()))?
1102 .to_string(),
1103 with: v["with"]
1104 .as_object()
1105 .map(|m| m.iter().map(|(k, v)| (k.clone(), v.clone())).collect())
1106 .unwrap_or_default(),
1107 }),
1108 Some(other) => return Err(at(format!("unknown op `{other}`"))),
1109 };
1110 Ok(Node { name, input, op })
1111}
1112
1113fn parse_guard(v: &Value, at: &dyn Fn(String) -> String) -> Result<Guard, String> {
1114 let map = v
1115 .as_object()
1116 .ok_or_else(|| at("`when` must be an object".into()))?;
1117 let mut tests = Vec::new();
1118 for (key, test) in map {
1119 let key = match key.strip_prefix('@') {
1120 Some(rest) => Key::System(rest.to_string()),
1121 None => Key::Field(key.clone()),
1122 };
1123 let test = match test {
1124 Value::Array(a) => Test::In(
1125 a.iter()
1126 .map(|v| {
1127 v.as_str()
1128 .map(String::from)
1129 .ok_or_else(|| at("a membership test takes strings".into()))
1130 })
1131 .collect::<Result<Vec<_>, String>>()?,
1132 ),
1133 Value::String(s) if s == "present" => Test::Present,
1134 Value::String(s) if s == "absent" => Test::Absent,
1135 Value::String(s) => {
1136 return Err(at(format!("`{s}` is not a test; use present or absent")));
1137 }
1138 Value::Object(o) if o.len() == 1 => {
1139 let (op, n) = o.iter().next().expect("one entry");
1140 let cmp =
1141 Cmp::parse(op).ok_or_else(|| at(format!("`{op}` is not a comparison")))?;
1142 Test::Compare(cmp, count(n, "a comparison").map_err(at)?)
1143 }
1144 _ => {
1145 return Err(at(
1146 "a test is a list, present, absent, or one comparison".into()
1147 ));
1148 }
1149 };
1150 tests.push((key, test));
1151 }
1152 Ok(Guard(tests))
1153}
1154
1155fn parse_edge(wf: &str, i: usize, v: &Value) -> Result<Edge, String> {
1156 let at = |e: String| format!("workflow `{wf}`: edge {}: {e}", i + 1);
1157 known(
1158 v,
1159 &["from", "to", "when", "default", "max_laps"],
1160 &format!("workflow `{wf}` edge {}", i + 1),
1161 )?;
1162 let from = match v["from"].as_str() {
1163 Some("@input") => From::Input,
1164 Some("@output") => return Err(at("`@output` is a destination, not a source".into())),
1165 Some(n) => From::Node(n.to_string()),
1166 None => return Err(at("no `from`".into())),
1167 };
1168 let to = match v["to"].as_str() {
1169 Some("@output") => To::Output,
1170 Some("@input") => return Err(at("`@input` is a source, not a destination".into())),
1171 Some(n) => To::Node(n.to_string()),
1172 None => return Err(at("no `to`".into())),
1173 };
1174 let when = match &v["when"] {
1175 Value::Null => None,
1176 g => Some(parse_guard(g, &at)?),
1177 };
1178 let default = v["default"].as_bool().unwrap_or(false);
1179 if default && when.is_some() {
1180 return Err(at("a default edge takes no `when`".into()));
1181 }
1182 Ok(Edge {
1183 from,
1184 to,
1185 when,
1186 default,
1187 max_laps: match &v["max_laps"] {
1188 Value::Null => None,
1189 c => Some(count(c, "max_laps").map_err(at)?),
1190 },
1191 })
1192}
1193
1194fn parse_workflow(v: &Value) -> Result<Workflow, String> {
1195 let name = v["name"]
1196 .as_str()
1197 .ok_or("a workflow with no `name`")?
1198 .to_string();
1199 let at = |e: String| format!("workflow `{name}`: {e}");
1200 known(
1201 v,
1202 &[
1203 "name", "in", "out", "effects", "params", "caps", "nodes", "edges",
1204 ],
1205 &format!("workflow `{name}`"),
1206 )?;
1207 let input = v["in"]
1208 .as_str()
1209 .ok_or_else(|| at("no `in` type".into()))?
1210 .to_string();
1211 let output = v["out"].as_str().map(String::from);
1212 let mut effects = Effects::default();
1213 if let Some(list) = v["effects"].as_array() {
1214 for e in list {
1215 match e.as_str() {
1216 Some("repo") => effects.repo = true,
1217 Some("writes") => effects.writes = true,
1218 other => {
1219 return Err(at(format!(
1220 "unknown effect `{}`; expected repo or writes",
1221 other.unwrap_or("")
1222 )));
1223 }
1224 }
1225 }
1226 }
1227 let params = match &v["params"] {
1228 Value::Null => BTreeMap::new(),
1229 p => parse_params(p, &at)?,
1230 };
1231 let caps = {
1232 let c = &v["caps"];
1233 known(
1234 c,
1235 &["max_units", "max_edits"],
1236 &format!("workflow `{name}` caps"),
1237 )?;
1238 Caps {
1239 max_units: c["max_units"]
1240 .as_i64()
1241 .ok_or_else(|| at("caps need `max_units`".into()))?,
1242 max_edits: c["max_edits"]
1243 .as_i64()
1244 .ok_or_else(|| at("caps need `max_edits`".into()))?,
1245 }
1246 };
1247 let nodes = v["nodes"]
1248 .as_array()
1249 .ok_or_else(|| at("no `nodes` array".into()))?
1250 .iter()
1251 .map(|n| parse_node(&name, n))
1252 .collect::<Result<Vec<_>, String>>()?;
1253 if nodes.is_empty() {
1254 return Err(at("a workflow with no nodes does nothing".into()));
1255 }
1256 let edges = v["edges"]
1257 .as_array()
1258 .ok_or_else(|| at("no `edges` array".into()))?
1259 .iter()
1260 .enumerate()
1261 .map(|(i, e)| parse_edge(&name, i, e))
1262 .collect::<Result<Vec<_>, String>>()?;
1263 Ok(Workflow {
1264 name,
1265 input,
1266 output,
1267 effects,
1268 params,
1269 caps,
1270 nodes,
1271 edges,
1272 })
1273}
1274
1275impl Document {
1278 fn validate(&self) -> Result<(), String> {
1279 let mut seen = BTreeSet::new();
1280 for w in &self.workflows {
1281 if !seen.insert(&w.name) {
1282 return Err(format!("workflow `{}` is declared twice", w.name));
1283 }
1284 }
1285 for w in &self.workflows {
1286 self.validate_workflow(w)?;
1287 }
1288 self.validate_calls()?;
1289 Ok(())
1290 }
1291
1292 fn output_type(&self, w: &Workflow, node: &Node) -> Result<String, String> {
1294 Ok(match &node.op {
1295 Op::Map(m) => m.emits.clone(),
1296 Op::Reduce(r) => r.emits.clone(),
1297 Op::Edit(_) | Op::Check { .. } | Op::Emit(_) => node.input.clone(),
1302 Op::Call(c) => {
1303 let callee = self.workflow(&c.workflow).ok_or_else(|| {
1304 format!(
1305 "workflow `{}`: node `{}`: unknown workflow `{}`",
1306 w.name, node.name, c.workflow
1307 )
1308 })?;
1309 callee.output.clone().ok_or_else(|| {
1310 format!(
1311 "workflow `{}`: node `{}`: `{}` returns nothing, so nothing may read it",
1312 w.name, node.name, c.workflow
1313 )
1314 })?
1315 }
1316 })
1317 }
1318
1319 fn validate_workflow(&self, w: &Workflow) -> Result<(), String> {
1320 let at = |e: String| format!("workflow `{}`: {e}", w.name);
1321 let node_at = |n: &str, e: String| format!("workflow `{}`: node `{n}`: {e}", w.name);
1322
1323 if self.resolve(&w.input).is_none() {
1324 return Err(at(format!("unknown input type `{}`", w.input)));
1325 }
1326 if let Some(out) = &w.output
1327 && self.resolve(out).is_none()
1328 {
1329 return Err(at(format!("unknown output type `{out}`")));
1330 }
1331 for (name, p) in &w.params {
1332 if let ParamType::Enum(values) = &p.ty
1333 && values.is_empty()
1334 {
1335 return Err(at(format!("parameter `{name}`: an enum with no values")));
1336 }
1337 }
1338
1339 let mut names = BTreeSet::new();
1340 for n in &w.nodes {
1341 if !names.insert(n.name.clone()) {
1342 return Err(at(format!("node `{}` is declared twice", n.name)));
1343 }
1344 if n.name.contains('/') {
1345 return Err(node_at(&n.name, "a node name may not contain `/`".into()));
1346 }
1347 if self.resolve(&n.input).is_none() {
1348 return Err(node_at(&n.name, format!("unknown type `{}`", n.input)));
1349 }
1350 self.validate_node(w, n)?;
1351 }
1352
1353 let mut pairs = BTreeSet::new();
1355 for (i, e) in w.edges.iter().enumerate() {
1356 let at = |msg: String| format!("workflow `{}`: edge {}: {msg}", w.name, i + 1);
1357 let from_type = match &e.from {
1358 From::Input => w.input.clone(),
1359 From::Node(n) => {
1360 let node = w.node(n).ok_or_else(|| at(format!("unknown node `{n}`")))?;
1361 self.output_type(w, node)?
1362 }
1363 };
1364 match &e.to {
1365 To::Output => {
1366 let out = w
1367 .output
1368 .as_deref()
1369 .ok_or_else(|| at("this workflow returns nothing".into()))?;
1370 if out != from_type {
1371 return Err(at(format!(
1372 "carries `{from_type}` into `@output`, which returns `{out}`"
1373 )));
1374 }
1375 }
1376 To::Node(n) => {
1377 let node = w.node(n).ok_or_else(|| at(format!("unknown node `{n}`")))?;
1378 if node.input != from_type {
1379 return Err(at(format!(
1380 "carries `{from_type}` into node `{n}`, which reads `{}`",
1381 node.input
1382 )));
1383 }
1384 }
1385 }
1386 let pair = (e.from.clone(), e.to.clone());
1387 if !pairs.insert(pair) {
1388 return Err(at(
1389 "that pair already has an edge; use one guard or two targets".into(),
1390 ));
1391 }
1392 if let Some(g) = &e.when {
1393 self.validate_guard(w, &from_type, g, &at)?;
1394 }
1395 if let Some(c) = &e.max_laps {
1396 self.bounded(w, c, &at, "max_laps")?;
1397 let From::Node(src) = &e.from else {
1398 return Err(at("a lap edge cannot start at `@input`".into()));
1399 };
1400 if !w.edges.iter().any(|o| o.from == e.from && o.default) {
1401 return Err(at(format!(
1402 "node `{src}` has a lap edge and no default edge, so a unit that \
1403 exhausts its laps has nowhere to go"
1404 )));
1405 }
1406 }
1407 }
1408
1409 self.validate_reachable(w)?;
1410 self.validate_acyclic(w)?;
1411 let effects = self.effects_of(w, &mut Vec::new())?;
1412 if effects != w.effects {
1413 let show = |e: Effects| match e.names().as_slice() {
1414 [] => "pure".to_string(),
1415 names => names.join(", "),
1416 };
1417 return Err(at(format!(
1418 "declares effects {}, but its graph has {}",
1419 show(w.effects),
1420 show(effects)
1421 )));
1422 }
1423 Ok(())
1424 }
1425
1426 fn bounded(
1429 &self,
1430 w: &Workflow,
1431 c: &Count,
1432 at: &dyn Fn(String) -> String,
1433 what: &str,
1434 ) -> Result<(), String> {
1435 match c {
1436 Count::Fixed(n) if *n < 0 => Err(at(format!("{what} cannot be negative"))),
1437 Count::Fixed(_) => Ok(()),
1438 Count::Param(p) => match w.params.get(p) {
1439 None => Err(at(format!("{what} names undeclared parameter `{p}`"))),
1440 Some(param) if param.max.is_none() => Err(at(format!(
1441 "{what} uses parameter `{p}`, which declares no `max`; a bound must be \
1442 knowable before a run"
1443 ))),
1444 Some(_) => Ok(()),
1445 },
1446 }
1447 }
1448
1449 fn validate_node(&self, w: &Workflow, n: &Node) -> Result<(), String> {
1450 let at = |e: String| format!("workflow `{}`: node `{}`: {e}", w.name, n.name);
1451 let self_edge = w
1452 .edges
1453 .iter()
1454 .any(|e| e.from == From::Node(n.name.clone()) && e.to == To::Node(n.name.clone()));
1455 let input = self.resolve(&n.input).expect("checked by the caller");
1456 match &n.op {
1457 Op::Map(m) => {
1458 if self.resolve(&m.emits).is_none() {
1459 return Err(at(format!("unknown type `{}`", m.emits)));
1460 }
1461 match m.via {
1462 Via::Agent if m.task.is_none() => {
1463 return Err(at("an agent node needs a `task`".into()));
1464 }
1465 Via::Rule if m.rule.is_none() => {
1466 return Err(at("a rule node needs a `rule`".into()));
1467 }
1468 _ => {}
1469 }
1470 if let Some(rule) = &m.rule {
1471 self.validate_map_rule(w, n, rule, &m.emits)?;
1472 }
1473 if m.out == Out::Grows {
1474 let units = m
1475 .max_units
1476 .as_ref()
1477 .ok_or_else(|| at("a `0..n` map must state `max_units`".into()))?;
1478 self.bounded(w, units, &at, "max_units")?;
1479 } else if !m.writes.is_empty() {
1480 let emits = self.resolve(&m.emits).expect("checked above");
1481 for f in &m.writes {
1482 if !emits.fields.contains_key(f) {
1483 return Err(at(format!(
1484 "`writes` names no field `{f}` of `{}`",
1485 m.emits
1486 )));
1487 }
1488 }
1489 }
1490 if self_edge {
1491 let depth = m
1492 .max_depth
1493 .as_ref()
1494 .ok_or_else(|| at("a self-edge must state `max_depth`".into()))?;
1495 self.bounded(w, depth, &at, "max_depth")?;
1496 if m.out != Out::Grows {
1497 return Err(at("only a `0..n` map may have a self-edge".into()));
1498 }
1499 }
1500 if let Some(t) = &m.task {
1501 self.placeholders(w, &input, t, &at)?;
1502 }
1503 if let Some(d) = &m.doc {
1504 self.placeholders(w, &input, d, &at)?;
1505 }
1506 if let Some(r) = &m.retry {
1507 self.bounded(w, &r.max, &at, "retry max")?;
1508 }
1509 }
1510 Op::Reduce(r) => {
1511 if self.resolve(&r.emits).is_none() {
1512 return Err(at(format!("unknown type `{}`", r.emits)));
1513 }
1514 match r.via {
1515 Via::Agent if r.task.is_none() => {
1516 return Err(at("an agent node needs a `task`".into()));
1517 }
1518 Via::Rule => {
1519 let rule = r
1520 .rule
1521 .as_deref()
1522 .ok_or_else(|| at("a rule node needs a `rule`".into()))?;
1523 if !REDUCE_RULES.contains(&rule)
1524 && !REDUCE_RULE_PREFIXES.iter().any(|p| rule.starts_with(p))
1525 {
1526 return Err(at(format!("unknown reduce rule `{rule}`")));
1527 }
1528 if let Some(n) = rule.strip_prefix("limit:")
1529 && !n.parse::<i64>().is_ok_and(|n| n >= 1)
1530 {
1531 return Err(at(format!(
1532 "rule `{rule}`: expected a count of 1 or more"
1533 )));
1534 }
1535 if rule == "rank" && !input.fields.contains_key("priority") {
1539 return Err(at(format!(
1540 "rule `rank` orders by `priority`, which `{}` does not declare",
1541 n.input
1542 )));
1543 }
1544 }
1545 _ => {}
1546 }
1547 for f in &r.group_by {
1548 if !input.fields.contains_key(f) {
1549 return Err(at(format!(
1550 "`group_by` names no field `{f}` of `{}`",
1551 n.input
1552 )));
1553 }
1554 }
1555 if let Some(t) = &r.task {
1556 self.placeholders(w, &input, t, &at)?;
1557 }
1558 }
1559 Op::Edit(e) => {
1560 self.placeholders(w, &input, &e.task, &at)?;
1561 if let Some(c) = &e.check
1562 && !CHECK_RULES.contains(&c.as_str())
1563 {
1564 return Err(at(format!("unknown check `{c}`")));
1565 }
1566 if let Some(r) = &e.retry {
1567 self.bounded(w, &r.max, &at, "retry max")?;
1568 }
1569 }
1570 Op::Check { rule } => {
1571 if !CHECK_RULES.contains(&rule.as_str()) {
1572 return Err(at(format!("unknown check rule `{rule}`")));
1573 }
1574 }
1575 Op::Emit(e) => {
1576 let (fields, needs): (&[&str], &[&str]) = match (e.sink, e.action) {
1579 (Sink::Todo, Action::Add) => {
1580 (&["text", "priority", "description"], &["text", "priority"])
1581 }
1582 (Sink::Todo, _) => (&["text"], &["text"]),
1583 (Sink::Note, Action::Add) => (&["text"], &["text"]),
1584 (Sink::Note, _) => return Err(at("a note is only added".into())),
1585 (Sink::Doc, _) => (&[], &[]),
1586 };
1587 for (field, placeholder) in &e.map {
1588 self.placeholders(w, &input, placeholder, &at)?;
1589 if e.sink != Sink::Doc && !fields.contains(&field.as_str()) {
1590 return Err(at(format!(
1591 "the sink writes no field `{field}`; it takes {}",
1592 fields.join(", ")
1593 )));
1594 }
1595 }
1596 if let Some(missing) = needs.iter().find(|f| !e.map.contains_key(**f)) {
1597 return Err(at(format!("this sink needs a `{missing}` mapping")));
1598 }
1599 }
1600 Op::Call(c) => {
1601 let callee = self
1602 .workflow(&c.workflow)
1603 .ok_or_else(|| at(format!("unknown workflow `{}`", c.workflow)))?;
1604 if callee.input != n.input {
1605 return Err(at(format!(
1606 "`{}` reads `{}`, but this node is given `{}`",
1607 c.workflow, callee.input, n.input
1608 )));
1609 }
1610 for (key, value) in &c.with {
1611 let param = callee.params.get(key).ok_or_else(|| {
1612 at(format!("`{}` declares no parameter `{key}`", c.workflow))
1613 })?;
1614 if let Some(p) = value.as_str().and_then(param_ref) {
1615 if !w.params.contains_key(&p) {
1616 return Err(at(format!("`{key}` names undeclared parameter `{p}`")));
1617 }
1618 } else if let ParamType::Enum(values) = ¶m.ty
1619 && !value
1620 .as_str()
1621 .is_some_and(|v| values.iter().any(|x| x == v))
1622 {
1623 return Err(at(format!(
1624 "`{key}` is `{value}`, which is not one of its values"
1625 )));
1626 } else if let (Some(max), Some(n)) = (param.max, value.as_i64())
1627 && n > max
1628 {
1629 return Err(at(format!(
1630 "`{key}` is {n}, over the maximum of {max} `{}` declares",
1631 c.workflow
1632 )));
1633 }
1634 }
1635 }
1636 }
1637 Ok(())
1638 }
1639
1640 fn validate_map_rule(
1641 &self,
1642 w: &Workflow,
1643 n: &Node,
1644 rule: &str,
1645 emits: &str,
1646 ) -> Result<(), String> {
1647 let at = |e: String| format!("workflow `{}`: node `{}`: {e}", w.name, n.name);
1648 let writes = match rule {
1651 "todo-items" => Some("item".to_string()),
1652 "open-runs" => Some("run".to_string()),
1653 "outdated-deps" => Some("signal".to_string()),
1654 _ if rule.starts_with("as:") => Some(rule["as:".len()..].to_string()),
1655 _ if rule.starts_with("where:") || rule.starts_with("path-") => Some(n.input.clone()),
1656 _ => None,
1657 };
1658 if let Some(ty) = writes
1659 && ty != emits
1660 {
1661 return Err(at(format!(
1662 "rule `{rule}` writes `{ty}`, and the node emits `{emits}`"
1663 )));
1664 }
1665 if MAP_RULES.contains(&rule) {
1666 return Ok(());
1667 }
1668 let Some(prefix) = MAP_RULE_PREFIXES.iter().find(|p| rule.starts_with(**p)) else {
1669 return Err(at(format!("unknown map rule `{rule}`")));
1670 };
1671 let arg = &rule[prefix.len()..];
1672 if arg.is_empty() {
1673 return Err(at(format!("rule `{rule}` names nothing after the colon")));
1674 }
1675 if *prefix == "as:" {
1676 let target = self
1677 .resolve(arg)
1678 .ok_or_else(|| at(format!("rule `{rule}`: unknown type `{arg}`")))?;
1679 let source = self.resolve(&n.input).expect("checked by the caller");
1680 for (field, decl) in &target.fields {
1681 if decl.required && !source.fields.contains_key(field) {
1682 return Err(at(format!(
1683 "rule `{rule}`: `{arg}` requires `{field}`, which `{}` does not have",
1684 n.input
1685 )));
1686 }
1687 }
1688 }
1689 Ok(())
1690 }
1691
1692 fn placeholders(
1696 &self,
1697 w: &Workflow,
1698 unit: &Type,
1699 text: &str,
1700 at: &dyn Fn(String) -> String,
1701 ) -> Result<(), String> {
1702 const LITERALS: [&str; 3] = ["in", "out", "doc"];
1703 let mut rest = text;
1704 while let Some(start) = rest.find('{') {
1705 let Some(end) = rest[start..].find('}') else {
1706 break;
1707 };
1708 let name = &rest[start + 1..start + end];
1709 rest = &rest[start + end + 1..];
1710 if name.is_empty() {
1711 continue;
1712 }
1713 if let Some(param) = name.strip_prefix('$') {
1714 if !w.params.contains_key(param) {
1715 return Err(at(format!("`{{${param}}}` names no parameter")));
1716 }
1717 } else if let Some(system) = name.strip_prefix('@') {
1718 if !SYSTEM_FIELDS.contains(&system) && !self.verdicts(w).contains(system) {
1719 return Err(at(format!(
1720 "`{{@{system}}}` names no system field or verdict"
1721 )));
1722 }
1723 } else if !LITERALS.contains(&name) && !unit.fields.contains_key(name) {
1724 return Err(at(format!(
1725 "`{{{name}}}` names no field of this node's input"
1726 )));
1727 }
1728 }
1729 Ok(())
1730 }
1731
1732 fn verdicts(&self, w: &Workflow) -> BTreeSet<String> {
1734 let mut out = BTreeSet::new();
1735 for n in &w.nodes {
1736 match &n.op {
1737 Op::Check { rule } => {
1738 out.insert(rule.clone());
1739 }
1740 Op::Edit(e) => {
1741 if let Some(c) = &e.check {
1742 out.insert(c.clone());
1743 }
1744 }
1745 Op::Call(c) => {
1746 if let Some(callee) = self.workflow(&c.workflow) {
1747 out.extend(self.verdicts(callee));
1748 }
1749 }
1750 _ => {}
1751 }
1752 }
1753 out
1754 }
1755
1756 fn validate_guard(
1757 &self,
1758 w: &Workflow,
1759 unit_type: &str,
1760 g: &Guard,
1761 at: &dyn Fn(String) -> String,
1762 ) -> Result<(), String> {
1763 let unit = self
1764 .resolve(unit_type)
1765 .ok_or_else(|| at(format!("unknown type `{unit_type}`")))?;
1766 for (key, test) in &g.0 {
1767 let field = match key {
1768 Key::System(s) => {
1769 if !SYSTEM_FIELDS.contains(&s.as_str()) && !self.verdicts(w).contains(s) {
1770 return Err(at(format!("`@{s}` names no system field or verdict")));
1771 }
1772 None
1773 }
1774 Key::Field(f) => Some(
1775 unit.fields
1776 .get(f)
1777 .ok_or_else(|| at(format!("`{f}` is no field of `{unit_type}`")))?,
1778 ),
1779 };
1780 match test {
1781 Test::In(values) => {
1782 for v in values {
1783 if let Some(p) = param_ref(v) {
1784 if !w.params.contains_key(&p) {
1785 return Err(at(format!("`{{${p}}}` names no parameter")));
1786 }
1787 continue;
1788 }
1789 if let Some(Field {
1790 ty: FieldType::Enum { values: allowed },
1791 ..
1792 }) = field
1793 && !allowed.iter().any(|a| a == v)
1794 {
1795 return Err(at(format!("`{v}` is not one of that enum's values")));
1796 }
1797 }
1798 }
1799 Test::Compare(_, c) => self.bounded(w, c, at, "a comparison")?,
1800 Test::Present | Test::Absent => {}
1801 }
1802 }
1803 Ok(())
1804 }
1805
1806 fn validate_reachable(&self, w: &Workflow) -> Result<(), String> {
1807 let mut seen: BTreeSet<&str> = BTreeSet::new();
1808 let mut queue: Vec<&str> = w
1809 .edges
1810 .iter()
1811 .filter(|e| e.from == From::Input)
1812 .filter_map(|e| match &e.to {
1813 To::Node(n) => Some(n.as_str()),
1814 To::Output => None,
1815 })
1816 .collect();
1817 if queue.is_empty() {
1818 return Err(format!(
1819 "workflow `{}`: no edge leaves `@input`, so nothing can run",
1820 w.name
1821 ));
1822 }
1823 while let Some(n) = queue.pop() {
1824 if !seen.insert(n) {
1825 continue;
1826 }
1827 for e in &w.edges {
1828 if e.from == From::Node(n.to_string())
1829 && let To::Node(next) = &e.to
1830 {
1831 queue.push(next.as_str());
1832 }
1833 }
1834 }
1835 match w.nodes.iter().find(|n| !seen.contains(n.name.as_str())) {
1836 Some(n) => Err(format!(
1837 "workflow `{}`: node `{}` is not reachable from `@input`",
1838 w.name, n.name
1839 )),
1840 None => Ok(()),
1841 }
1842 }
1843
1844 fn validate_acyclic(&self, w: &Workflow) -> Result<(), String> {
1848 let forward: Vec<&Edge> = w.edges.iter().filter(|e| e.is_forward()).collect();
1849 let mut indegree: BTreeMap<&str, usize> =
1850 w.nodes.iter().map(|n| (n.name.as_str(), 0)).collect();
1851 for e in &forward {
1852 if let (From::Node(_), To::Node(to)) = (&e.from, &e.to)
1853 && let Some(d) = indegree.get_mut(to.as_str())
1854 {
1855 *d += 1;
1856 }
1857 }
1858 let mut ready: Vec<&str> = indegree
1859 .iter()
1860 .filter(|(_, d)| **d == 0)
1861 .map(|(n, _)| *n)
1862 .collect();
1863 let mut settled = 0;
1864 while let Some(n) = ready.pop() {
1865 settled += 1;
1866 for e in &forward {
1867 if e.from == From::Node(n.to_string())
1868 && let To::Node(to) = &e.to
1869 && let Some(d) = indegree.get_mut(to.as_str())
1870 {
1871 *d -= 1;
1872 if *d == 0 {
1873 ready.push(to.as_str());
1874 }
1875 }
1876 }
1877 }
1878 if settled < w.nodes.len() {
1879 return Err(format!(
1880 "workflow `{}`: a cycle that is not a `max_laps` back edge",
1881 w.name
1882 ));
1883 }
1884 Ok(())
1885 }
1886
1887 fn effects_of(&self, w: &Workflow, stack: &mut Vec<String>) -> Result<Effects, String> {
1889 if stack.contains(&w.name) {
1890 return Err(format!(
1891 "workflow `{}` calls itself, directly or through another; \
1892 a call is resolved before a run, so it could not terminate",
1893 w.name
1894 ));
1895 }
1896 stack.push(w.name.clone());
1897 let mut effects = Effects::default();
1898 for n in &w.nodes {
1899 effects = effects.union(match &n.op {
1900 Op::Edit(_) => Effects {
1901 repo: true,
1902 writes: false,
1903 },
1904 Op::Emit(_) => Effects {
1905 repo: false,
1906 writes: true,
1907 },
1908 Op::Call(c) => {
1909 let callee = self.workflow(&c.workflow).ok_or_else(|| {
1910 format!("workflow `{}`: unknown `{}`", w.name, c.workflow)
1911 })?;
1912 self.effects_of(callee, stack)?
1913 }
1914 _ => Effects::default(),
1915 });
1916 }
1917 stack.pop();
1918 Ok(effects)
1919 }
1920
1921 fn validate_calls(&self) -> Result<(), String> {
1922 for w in &self.workflows {
1923 self.effects_of(w, &mut Vec::new())?;
1924 }
1925 Ok(())
1926 }
1927}
1928
1929pub fn holds(
1938 guard: &Guard,
1939 data: &Value,
1940 system: &std::collections::BTreeMap<String, Value>,
1941) -> std::result::Result<(), String> {
1942 for (key, test) in &guard.0 {
1943 let (name, value) = match key {
1944 Key::Field(f) => (f.clone(), data.get(f).cloned()),
1945 Key::System(s) => (format!("@{s}"), system.get(s).cloned()),
1946 };
1947 let text = value.as_ref().and_then(|v| match v {
1948 Value::String(s) => Some(s.clone()),
1949 Value::Null => None,
1950 other => Some(other.to_string()),
1951 });
1952 match test {
1953 Test::In(want) => {
1954 let held = text.as_deref().is_some_and(|t| want.iter().any(|w| w == t));
1955 if !held {
1956 return Err(format!(
1957 "{name} is {}, not {}",
1958 text.as_deref().unwrap_or("unset"),
1959 want.join(" or ")
1960 ));
1961 }
1962 }
1963 Test::Compare(cmp, Count::Fixed(n)) => {
1964 let Some(have) = text.as_deref().and_then(|t| t.parse::<i64>().ok()) else {
1965 return Err(format!("{name} is not a number"));
1966 };
1967 let held = match cmp {
1968 Cmp::Eq => have == *n,
1969 Cmp::Lt => have < *n,
1970 Cmp::Le => have <= *n,
1971 Cmp::Gt => have > *n,
1972 Cmp::Ge => have >= *n,
1973 };
1974 if !held {
1975 return Err(format!("{name} is {have}"));
1976 }
1977 }
1978 Test::Compare(_, Count::Param(p)) => {
1980 return Err(format!("{name}: parameter `{p}` was not substituted"));
1981 }
1982 Test::Present => {
1983 if text.is_none() {
1984 return Err(format!("{name} is unset"));
1985 }
1986 }
1987 Test::Absent => {
1988 if text.is_some() {
1989 return Err(format!("{name} is set"));
1990 }
1991 }
1992 }
1993 }
1994 Ok(())
1995}
1996
1997impl Count {
2000 fn to_json(&self) -> Value {
2001 match self {
2002 Count::Fixed(n) => Value::from(*n),
2003 Count::Param(p) => Value::from(format!("{{${p}}}")),
2004 }
2005 }
2006}
2007
2008impl FieldType {
2009 fn write(&self, m: &mut serde_json::Map<String, Value>) {
2010 match self {
2011 FieldType::Line { max, unique } => {
2012 m.insert("type".into(), "line".into());
2013 m.insert("max".into(), Value::from(*max));
2014 if *unique {
2015 m.insert("unique".into(), "normalised".into());
2016 }
2017 }
2018 FieldType::Lines { max } => {
2019 m.insert("type".into(), "lines".into());
2020 m.insert("max".into(), Value::from(*max));
2021 }
2022 FieldType::Enum { values } => {
2023 m.insert("type".into(), "enum".into());
2024 m.insert("values".into(), values.clone().into());
2025 }
2026 FieldType::List => {
2027 m.insert("type".into(), "list".into());
2028 }
2029 FieldType::Int { min, max } => {
2030 m.insert("type".into(), "int".into());
2031 if let Some(n) = min {
2032 m.insert("min".into(), Value::from(*n));
2033 }
2034 if let Some(n) = max {
2035 m.insert("max".into(), Value::from(*n));
2036 }
2037 }
2038 FieldType::Bool => {
2039 m.insert("type".into(), "bool".into());
2040 }
2041 }
2042 }
2043}
2044
2045impl Guard {
2046 fn to_json(&self) -> Value {
2047 let mut m = serde_json::Map::new();
2048 for (key, test) in &self.0 {
2049 let name = match key {
2050 Key::Field(f) => f.clone(),
2051 Key::System(s) => format!("@{s}"),
2052 };
2053 m.insert(
2054 name,
2055 match test {
2056 Test::In(values) => values.clone().into(),
2057 Test::Compare(cmp, c) => {
2058 let op = match cmp {
2059 Cmp::Eq => "==",
2060 Cmp::Lt => "<",
2061 Cmp::Le => "<=",
2062 Cmp::Gt => ">",
2063 Cmp::Ge => ">=",
2064 };
2065 serde_json::json!({op: c.to_json()})
2066 }
2067 Test::Present => "present".into(),
2068 Test::Absent => "absent".into(),
2069 },
2070 );
2071 }
2072 Value::Object(m)
2073 }
2074}
2075
2076impl Document {
2077 pub fn to_json(&self) -> String {
2082 let types: serde_json::Map<String, Value> = self
2083 .types
2084 .iter()
2085 .map(|(name, ty)| {
2086 let fields: serde_json::Map<String, Value> = ty
2087 .fields
2088 .iter()
2089 .map(|(field, decl)| {
2090 let mut m = serde_json::Map::new();
2091 decl.ty.write(&mut m);
2092 if decl.required {
2093 m.insert("required".into(), true.into());
2094 }
2095 (field.clone(), Value::Object(m))
2096 })
2097 .collect();
2098 (
2099 name.clone(),
2100 serde_json::json!({"fields": Value::Object(fields)}),
2101 )
2102 })
2103 .collect();
2104 let workflows: Vec<Value> = self.workflows.iter().map(Workflow::to_json).collect();
2105 let mut doc = serde_json::Map::new();
2106 if !types.is_empty() {
2107 doc.insert("types".into(), Value::Object(types));
2108 }
2109 doc.insert("workflow".into(), Value::Array(workflows));
2110 serde_json::to_string_pretty(&Value::Object(doc)).unwrap_or_default()
2111 }
2112}
2113
2114impl Workflow {
2115 fn to_json(&self) -> Value {
2116 let params: serde_json::Map<String, Value> = self
2117 .params
2118 .iter()
2119 .map(|(name, p)| {
2120 let mut m = serde_json::Map::new();
2121 m.insert(
2122 "type".into(),
2123 match &p.ty {
2124 ParamType::Name => "name",
2125 ParamType::Line => "line",
2126 ParamType::Int => "int",
2127 ParamType::Enum(_) => "enum",
2128 ParamType::Bool => "bool",
2129 ParamType::List => "list",
2130 }
2131 .into(),
2132 );
2133 if let ParamType::Enum(values) = &p.ty {
2134 m.insert("values".into(), values.clone().into());
2135 }
2136 m.insert("default".into(), p.default.clone());
2137 if let Some(max) = p.max {
2138 m.insert("max".into(), Value::from(max));
2139 }
2140 (name.clone(), Value::Object(m))
2141 })
2142 .collect();
2143 let mut w = serde_json::Map::new();
2144 w.insert("name".into(), self.name.clone().into());
2145 w.insert("in".into(), self.input.clone().into());
2146 if let Some(out) = &self.output {
2147 w.insert("out".into(), out.clone().into());
2148 }
2149 w.insert("effects".into(), self.effects.names().into());
2150 w.insert("params".into(), Value::Object(params));
2151 w.insert(
2152 "caps".into(),
2153 serde_json::json!({"max_units": self.caps.max_units, "max_edits": self.caps.max_edits}),
2154 );
2155 w.insert(
2156 "nodes".into(),
2157 self.nodes
2158 .iter()
2159 .map(Node::to_json)
2160 .collect::<Vec<_>>()
2161 .into(),
2162 );
2163 w.insert(
2164 "edges".into(),
2165 self.edges
2166 .iter()
2167 .map(Edge::to_json)
2168 .collect::<Vec<_>>()
2169 .into(),
2170 );
2171 Value::Object(w)
2172 }
2173}
2174
2175impl Retry {
2176 fn to_json(&self) -> Value {
2177 let mut m = serde_json::Map::new();
2178 m.insert("max".into(), self.max.to_json());
2179 m.insert("while".into(), self.predicate.clone().into());
2180 if let Some(model) = &self.escalate {
2181 m.insert("escalate".into(), serde_json::json!({"model": model}));
2182 }
2183 Value::Object(m)
2184 }
2185}
2186
2187impl Node {
2188 fn to_json(&self) -> Value {
2189 let mut m = serde_json::Map::new();
2190 m.insert("name".into(), self.name.clone().into());
2191 m.insert("op".into(), self.op.name().into());
2192 m.insert("in".into(), self.input.clone().into());
2193 let via = |m: &mut serde_json::Map<String, Value>, via: Via| {
2194 m.insert(
2195 "via".into(),
2196 match via {
2197 Via::Agent => "agent",
2198 Via::Rule => "rule",
2199 }
2200 .into(),
2201 );
2202 };
2203 match &self.op {
2204 Op::Map(n) => {
2205 via(&mut m, n.via);
2206 m.insert("out".into(), n.out.name().into());
2207 m.insert("emits".into(), n.emits.clone().into());
2208 if !n.writes.is_empty() {
2209 m.insert("writes".into(), n.writes.clone().into());
2210 }
2211 if let Some(c) = &n.max_units {
2212 m.insert("max_units".into(), c.to_json());
2213 }
2214 if let Some(c) = &n.max_depth {
2215 m.insert("max_depth".into(), c.to_json());
2216 }
2217 if let Some(s) = &n.task {
2218 m.insert("task".into(), s.clone().into());
2219 }
2220 if let Some(s) = &n.rule {
2221 m.insert("rule".into(), s.clone().into());
2222 }
2223 if let Some(s) = &n.doc {
2224 m.insert("doc".into(), s.clone().into());
2225 }
2226 if n.publish {
2227 m.insert("publish".into(), true.into());
2228 }
2229 if let Some(r) = &n.retry {
2230 m.insert("retry".into(), r.to_json());
2231 }
2232 }
2233 Op::Reduce(n) => {
2234 via(&mut m, n.via);
2235 m.insert("emits".into(), n.emits.clone().into());
2236 if !n.group_by.is_empty() {
2237 m.insert("group_by".into(), n.group_by.clone().into());
2238 }
2239 if let Some(s) = &n.task {
2240 m.insert("task".into(), s.clone().into());
2241 }
2242 if let Some(s) = &n.rule {
2243 m.insert("rule".into(), s.clone().into());
2244 }
2245 }
2246 Op::Edit(n) => {
2247 m.insert("task".into(), n.task.clone().into());
2248 if let Some(c) = &n.check {
2249 m.insert("check".into(), c.clone().into());
2250 }
2251 if let Some(r) = &n.retry {
2252 m.insert("retry".into(), r.to_json());
2253 }
2254 }
2255 Op::Check { rule } => {
2256 m.insert("rule".into(), rule.clone().into());
2257 }
2258 Op::Emit(n) => {
2259 m.insert(
2260 "sink".into(),
2261 match n.sink {
2262 Sink::Todo => "todo",
2263 Sink::Note => "note",
2264 Sink::Doc => "doc",
2265 }
2266 .into(),
2267 );
2268 m.insert(
2269 "action".into(),
2270 match n.action {
2271 Action::Add => "add",
2272 Action::Tick => "tick",
2273 Action::Remove => "remove",
2274 }
2275 .into(),
2276 );
2277 m.insert(
2278 "map".into(),
2279 Value::Object(
2280 n.map
2281 .iter()
2282 .map(|(k, v)| (k.clone(), Value::from(v.clone())))
2283 .collect(),
2284 ),
2285 );
2286 }
2287 Op::Call(n) => {
2288 m.insert("workflow".into(), n.workflow.clone().into());
2289 m.insert(
2290 "with".into(),
2291 Value::Object(n.with.iter().map(|(k, v)| (k.clone(), v.clone())).collect()),
2292 );
2293 }
2294 }
2295 Value::Object(m)
2296 }
2297}
2298
2299impl Edge {
2300 fn to_json(&self) -> Value {
2301 let mut m = serde_json::Map::new();
2302 m.insert(
2303 "from".into(),
2304 match &self.from {
2305 From::Input => "@input".to_string(),
2306 From::Node(n) => n.clone(),
2307 }
2308 .into(),
2309 );
2310 m.insert(
2311 "to".into(),
2312 match &self.to {
2313 To::Output => "@output".to_string(),
2314 To::Node(n) => n.clone(),
2315 }
2316 .into(),
2317 );
2318 if let Some(g) = &self.when {
2319 m.insert("when".into(), g.to_json());
2320 }
2321 if self.default {
2322 m.insert("default".into(), true.into());
2323 }
2324 if let Some(c) = &self.max_laps {
2325 m.insert("max_laps".into(), c.to_json());
2326 }
2327 Value::Object(m)
2328 }
2329}
2330
2331#[derive(Debug, Clone, PartialEq)]
2335pub struct NodeBound {
2336 pub node: String,
2337 pub units_in: i64,
2339 pub units_out: i64,
2341 pub runs: i64,
2343 pub agent_runs: i64,
2345}
2346
2347impl Document {
2348 pub fn flatten(&self, workflow: &str) -> Result<Workflow, String> {
2356 let mut flat = self
2357 .workflow(workflow)
2358 .ok_or_else(|| format!("unknown workflow `{workflow}`"))?
2359 .clone();
2360 while flat.nodes.iter().any(|n| matches!(n.op, Op::Call(_))) {
2363 flat = self.inline(&flat)?;
2364 }
2365 check_flat(&flat)?;
2366 Ok(flat)
2367 }
2368
2369 pub fn unbuilt(&self) -> Vec<String> {
2372 let mut out = Vec::new();
2373 for w in &self.workflows {
2374 let at = |what: String| format!("workflow `{}`: {what}", w.name);
2375 for (i, e) in w.edges.iter().enumerate() {
2376 if e.max_laps.is_some() {
2377 out.push(at(format!("edge {}: a lap edge is not taken yet", i + 1)));
2378 } else if !e.is_forward() {
2379 out.push(at(format!(
2380 "edge {}: recursion (a self-edge) is not taken yet",
2381 i + 1
2382 )));
2383 }
2384 let reads_children = e.when.as_ref().is_some_and(|g| {
2385 g.0.iter()
2386 .any(|(k, _)| *k == Key::System("children".into()))
2387 });
2388 if reads_children {
2389 out.push(at(format!(
2390 "edge {}: `@children` is not recorded yet",
2391 i + 1
2392 )));
2393 }
2394 }
2395 for n in &w.nodes {
2396 let at = |what: &str| at(format!("node `{}`: {what}", n.name));
2397 let texts: Vec<&String> = match &n.op {
2398 Op::Map(m) => m.task.iter().chain(m.doc.iter()).collect(),
2399 Op::Reduce(r) => r.task.iter().collect(),
2400 Op::Edit(e) => vec![&e.task],
2401 Op::Emit(e) => e.map.values().collect(),
2402 _ => Vec::new(),
2403 };
2404 if texts.iter().any(|t| t.contains("{@children}")) {
2405 out.push(at("`@children` is not recorded yet"));
2406 }
2407 match &n.op {
2408 Op::Map(m) if m.retry.is_some() => out.push(at("`retry` is not taken yet")),
2409 Op::Edit(e) if e.retry.is_some() => out.push(at("`retry` is not taken yet")),
2410 _ => {}
2411 }
2412 match &n.op {
2413 Op::Map(m) if m.publish => out.push(at("`publish` is not built yet")),
2414 Op::Emit(e) if e.sink == Sink::Doc => {
2415 out.push(at("the `doc` sink is not built yet"))
2416 }
2417 Op::Check { rule } if rule == "nonempty" => out.push(at(
2418 "`nonempty` is not built: a check writes a verdict per unit, and an empty bag has none",
2419 )),
2420 Op::Edit(EditNode { check: Some(c), .. }) if c == "nonempty" => {
2421 out.push(at("`nonempty` is not built"))
2422 }
2423 _ => {}
2424 }
2425 }
2426 }
2427 out
2428 }
2429
2430 fn inline(&self, w: &Workflow) -> Result<Workflow, String> {
2432 let site = w
2433 .nodes
2434 .iter()
2435 .find(|n| matches!(n.op, Op::Call(_)))
2436 .expect("checked by the caller");
2437 let Op::Call(c) = &site.op else {
2438 unreachable!()
2439 };
2440 let at = |e: String| format!("workflow `{}`: node `{}`: {e}", w.name, site.name);
2441 let callee = self
2442 .workflow(&c.workflow)
2443 .ok_or_else(|| at(format!("unknown workflow `{}`", c.workflow)))?;
2444
2445 let mut out = w.clone();
2449 let mut rename: BTreeMap<String, String> = BTreeMap::new();
2450 for (p, decl) in &callee.params {
2451 let passed = c.with.get(p);
2452 match passed.and_then(Value::as_str).and_then(param_ref) {
2453 Some(q) => {
2454 if decl.max.is_some()
2455 && w.params.get(&q).is_some_and(|caller| caller.max.is_none())
2456 {
2457 return Err(at(format!(
2458 "`{p}` bounds `{}` and takes `{q}`, which declares no `max`",
2459 c.workflow
2460 )));
2461 }
2462 rename.insert(p.clone(), q);
2463 }
2464 None => {
2465 let qualified = format!("{}/{p}", site.name);
2466 out.params.insert(
2467 qualified.clone(),
2468 Param {
2469 ty: decl.ty.clone(),
2470 default: passed.cloned().unwrap_or_else(|| decl.default.clone()),
2471 max: match passed.and_then(Value::as_i64) {
2475 Some(n) if decl.max.is_some() => Some(n),
2476 _ => decl.max,
2477 },
2478 },
2479 );
2480 rename.insert(p.clone(), qualified);
2481 }
2482 }
2483 }
2484
2485 let qualify = |name: &str| format!("{}/{name}", site.name);
2486 out.nodes.retain(|n| n.name != site.name);
2487 for n in &callee.nodes {
2488 let mut n = n.clone();
2489 n.name = qualify(&n.name);
2490 rename_node(&mut n, &rename);
2491 out.nodes.push(n);
2492 }
2493
2494 let into: Vec<Edge> = w
2495 .edges
2496 .iter()
2497 .filter(|e| e.to == To::Node(site.name.clone()))
2498 .cloned()
2499 .collect();
2500 let from: Vec<Edge> = w
2501 .edges
2502 .iter()
2503 .filter(|e| e.from == From::Node(site.name.clone()))
2504 .cloned()
2505 .collect();
2506 out.edges.retain(|e| {
2507 e.to != To::Node(site.name.clone()) && e.from != From::Node(site.name.clone())
2508 });
2509
2510 for e in &callee.edges {
2511 let mut e = e.clone();
2512 rename_guard(&mut e, &rename);
2513 match (&e.from, &e.to) {
2514 (From::Input, To::Node(to)) => {
2517 for i in &into {
2518 out.edges
2519 .push(join(&at, i, &e, i.from.clone(), To::Node(qualify(to)))?);
2520 }
2521 }
2522 (From::Node(src), To::Output) => {
2523 for o in &from {
2524 out.edges
2525 .push(join(&at, &e, o, From::Node(qualify(src)), o.to.clone())?);
2526 }
2527 }
2528 (From::Input, To::Output) => {
2532 return Err(at(format!(
2533 "`{}` passes its input straight out",
2534 c.workflow
2535 )));
2536 }
2537 (From::Node(src), To::Node(to)) => {
2538 e.from = From::Node(qualify(src));
2539 e.to = To::Node(qualify(to));
2540 out.edges.push(e);
2541 }
2542 }
2543 }
2544 Ok(out)
2545 }
2546}
2547
2548fn check_flat(w: &Workflow) -> Result<(), String> {
2552 let mut pairs = BTreeSet::new();
2553 let mut defaults = BTreeSet::new();
2554 for e in &w.edges {
2555 let from = match &e.from {
2556 From::Input => "@input".to_string(),
2557 From::Node(n) => n.clone(),
2558 };
2559 if !pairs.insert((e.from.clone(), e.to.clone())) {
2560 return Err(format!(
2561 "workflow `{}`: after calls are inlined, `{from}` has two edges to one target",
2562 w.name
2563 ));
2564 }
2565 if e.default && !defaults.insert(from.clone()) {
2566 return Err(format!(
2567 "workflow `{}`: after calls are inlined, `{from}` has two default edges",
2568 w.name
2569 ));
2570 }
2571 }
2572 Ok(())
2573}
2574
2575fn join(
2579 at: &dyn Fn(String) -> String,
2580 first: &Edge,
2581 second: &Edge,
2582 from: From,
2583 to: To,
2584) -> Result<Edge, String> {
2585 let when = match (&first.when, &second.when) {
2586 (None, g) | (g, None) => g.clone(),
2587 (Some(a), Some(b)) => {
2588 let mut merged = a.clone();
2589 for (key, test) in &b.0 {
2590 match merged.0.iter().find(|(k, _)| k == key) {
2591 Some((_, held)) if held != test => {
2592 return Err(at(format!(
2593 "the call site and `{}` both guard the same field, differently",
2594 match key {
2595 Key::Field(f) => f.clone(),
2596 Key::System(f) => format!("@{f}"),
2597 }
2598 )));
2599 }
2600 Some(_) => {}
2601 None => merged.0.push((key.clone(), test.clone())),
2602 }
2603 }
2604 Some(merged)
2605 }
2606 };
2607 let max_laps = match (&first.max_laps, &second.max_laps) {
2608 (Some(_), Some(_)) => {
2609 return Err(at(
2610 "a lap edge on both sides of a call cannot be composed".into()
2611 ));
2612 }
2613 (a, b) => a.clone().or_else(|| b.clone()),
2614 };
2615 Ok(Edge {
2616 from,
2617 to,
2618 when,
2619 default: first.default || second.default,
2620 max_laps,
2621 })
2622}
2623
2624fn rename_node(n: &mut Node, rename: &BTreeMap<String, String>) {
2627 let text = |s: &mut String| rename_text(s, rename);
2628 match &mut n.op {
2629 Op::Map(m) => {
2630 for c in [&mut m.max_units, &mut m.max_depth].into_iter().flatten() {
2631 rename_count(c, rename);
2632 }
2633 for s in [&mut m.task, &mut m.doc].into_iter().flatten() {
2634 text(s);
2635 }
2636 if let Some(r) = &mut m.retry {
2637 rename_retry(r, rename);
2638 }
2639 }
2640 Op::Reduce(r) => {
2641 if let Some(t) = &mut r.task {
2642 text(t);
2643 }
2644 }
2645 Op::Edit(e) => {
2646 text(&mut e.task);
2647 if let Some(r) = &mut e.retry {
2648 rename_retry(r, rename);
2649 }
2650 }
2651 Op::Emit(e) => {
2652 for v in e.map.values_mut() {
2653 text(v);
2654 }
2655 }
2656 Op::Call(c) => {
2659 for v in c.with.values_mut() {
2660 if let Some(s) = v.as_str() {
2661 let mut owned = s.to_string();
2662 rename_text(&mut owned, rename);
2663 *v = Value::from(owned);
2664 }
2665 }
2666 }
2667 Op::Check { .. } => {}
2668 }
2669}
2670
2671fn rename_retry(r: &mut Retry, rename: &BTreeMap<String, String>) {
2672 rename_count(&mut r.max, rename);
2673 rename_text(&mut r.predicate, rename);
2674}
2675
2676fn rename_count(c: &mut Count, rename: &BTreeMap<String, String>) {
2677 if let Count::Param(p) = c
2678 && let Some(to) = rename.get(p)
2679 {
2680 *p = to.clone();
2681 }
2682}
2683
2684fn rename_text(s: &mut String, rename: &BTreeMap<String, String>) {
2687 let mut out = String::with_capacity(s.len());
2688 let mut rest = s.as_str();
2689 while let Some(start) = rest.find("{$") {
2690 out.push_str(&rest[..start]);
2691 let after = &rest[start + 2..];
2692 match after.find('}') {
2693 Some(end) => {
2694 let name = &after[..end];
2695 out.push_str("{$");
2696 out.push_str(rename.get(name).map_or(name, String::as_str));
2697 out.push('}');
2698 rest = &after[end + 1..];
2699 }
2700 None => {
2701 out.push_str(&rest[start..]);
2702 rest = "";
2703 }
2704 }
2705 }
2706 out.push_str(rest);
2707 *s = out;
2708}
2709
2710fn rename_guard(e: &mut Edge, rename: &BTreeMap<String, String>) {
2711 if let Some(c) = &mut e.max_laps {
2712 rename_count(c, rename);
2713 }
2714 let Some(g) = &mut e.when else { return };
2715 for (_, test) in &mut g.0 {
2716 match test {
2717 Test::In(values) => {
2718 for v in values {
2719 rename_text(v, rename);
2720 }
2721 }
2722 Test::Compare(_, c) => rename_count(c, rename),
2723 Test::Present | Test::Absent => {}
2724 }
2725 }
2726}
2727
2728impl Workflow {
2729 pub fn bind(&self, set: &[String]) -> Result<BTreeMap<String, Value>, String> {
2733 let mut out = BTreeMap::new();
2734 for arg in set {
2735 let (name, text) = arg
2736 .split_once('=')
2737 .ok_or_else(|| format!("`{arg}`: expected name=value"))?;
2738 if let Some((site, _)) = name.split_once('/') {
2741 return Err(format!(
2742 "`{name}` is set by the call site `{site}`, not on the command line"
2743 ));
2744 }
2745 let own: Vec<&String> = self.params.keys().filter(|k| !k.contains('/')).collect();
2746 let param = self.params.get(name).ok_or_else(|| {
2747 format!(
2748 "`{}` declares no parameter `{name}`; it takes {}",
2749 self.name,
2750 match own.is_empty() {
2751 true => "none".to_string(),
2752 false => own
2753 .iter()
2754 .map(|k| k.as_str())
2755 .collect::<Vec<_>>()
2756 .join(", "),
2757 }
2758 )
2759 })?;
2760 let value = param
2761 .ty
2762 .read(text)
2763 .map_err(|e| format!("parameter `{name}`: {e}"))?;
2764 if let (ParamType::Int, Some(max)) = (¶m.ty, param.max)
2765 && value.as_i64().is_some_and(|n| n > max)
2766 {
2767 return Err(format!(
2768 "parameter `{name}`: {text} is over its declared maximum of {max}"
2769 ));
2770 }
2771 out.insert(name.to_string(), value);
2772 }
2773 Ok(out)
2774 }
2775
2776 pub fn arguments(&self, given: &Value) -> BTreeMap<String, Value> {
2779 let mut out: BTreeMap<String, Value> = self
2780 .params
2781 .iter()
2782 .map(|(k, p)| (k.clone(), p.default.clone()))
2783 .collect();
2784 if let Some(map) = given.as_object() {
2785 for (k, v) in map {
2786 if self.params.contains_key(k) {
2787 out.insert(k.clone(), v.clone());
2788 }
2789 }
2790 }
2791 out
2792 }
2793}
2794
2795#[derive(Debug, Clone, PartialEq)]
2796pub struct Estimate {
2797 pub workflow: String,
2798 pub per_node: Vec<NodeBound>,
2799 pub runs: i64,
2800 pub agent_runs: i64,
2801 pub edits: i64,
2802 pub cost: f64,
2803}
2804
2805fn times(a: i64, b: i64) -> i64 {
2809 a.saturating_mul(b)
2810}
2811
2812impl Document {
2813 pub fn estimate(
2817 &self,
2818 workflow: &str,
2819 input_units: i64,
2820 agent_budget: f64,
2821 ) -> Result<Estimate, String> {
2822 self.estimate_inner(&self.flatten(workflow)?, input_units, agent_budget)
2825 }
2826
2827 fn estimate_inner(
2828 &self,
2829 w: &Workflow,
2830 input_units: i64,
2831 agent_budget: f64,
2832 ) -> Result<Estimate, String> {
2833 let order = self.topological(w);
2834 let mut out_units: BTreeMap<&str, i64> = BTreeMap::new();
2835 let mut per_node = Vec::new();
2836 let (mut runs, mut agent_runs, mut edits, mut cost) = (0i64, 0i64, 0i64, 0.0);
2837
2838 for name in &order {
2839 let node = w.node(name).expect("from the node list");
2840 let mut units_in = 0i64;
2841 for e in &w.edges {
2842 if let To::Node(to) = &e.to
2843 && to == name
2844 {
2845 units_in = units_in.saturating_add(match &e.from {
2846 From::Input => input_units,
2847 From::Node(src) => *out_units.get(src.as_str()).unwrap_or(&0),
2848 });
2849 }
2850 }
2851 units_in = units_in.min(w.caps.max_units);
2852
2853 let laps: i64 = w
2854 .edges
2855 .iter()
2856 .filter(|e| matches!(&e.to, To::Node(to) if to == name))
2857 .filter_map(|e| e.max_laps.as_ref())
2858 .map(|c| c.ceiling(&w.params))
2859 .sum();
2860 let retry = match &node.op {
2861 Op::Map(m) => m.retry.as_ref(),
2862 Op::Edit(e) => e.retry.as_ref(),
2863 _ => None,
2864 }
2865 .map_or(0, |r| r.max.ceiling(&w.params));
2866
2867 let recursive = w
2874 .edges
2875 .iter()
2876 .any(|e| e.from == From::Node(name.clone()) && e.to == To::Node(name.clone()));
2877 let mut units_seen = times(units_in, 1 + laps);
2878 if recursive && let Op::Map(m) = &node.op {
2879 let width = m.max_units.as_ref().map_or(0, |c| c.ceiling(&w.params));
2880 let depth = m.max_depth.as_ref().map_or(0, |c| c.ceiling(&w.params));
2881 let (mut level, mut total) = (units_seen, units_seen);
2882 for _ in 0..depth {
2883 level = times(level, width);
2884 total = total.saturating_add(level);
2885 }
2886 units_seen = total.min(units_seen.saturating_add(w.caps.max_units));
2887 }
2888 let node_runs = times(units_seen, 1 + retry);
2889 let (units_out, node_agent_runs, node_edits, node_cost) = match &node.op {
2890 Op::Map(m) => {
2891 let out = match m.out {
2892 Out::Grows => {
2893 let width = m.max_units.as_ref().map_or(0, |c| c.ceiling(&w.params));
2894 let depth = match recursive {
2897 true => 0,
2898 false => m.max_depth.as_ref().map_or(0, |c| c.ceiling(&w.params)),
2899 };
2900 let mut total = 0i64;
2903 let mut level = units_seen;
2904 for _ in 0..=depth {
2905 level = times(level, width);
2906 total = total.saturating_add(level);
2907 }
2908 total
2909 }
2910 Out::Same | Out::Shrinks => units_seen,
2911 };
2912 let agent = matches!(m.via, Via::Agent);
2913 (
2914 out.min(w.caps.max_units),
2915 if agent { node_runs } else { 0 },
2916 0,
2917 if agent {
2918 node_runs as f64 * agent_budget
2919 } else {
2920 0.0
2921 },
2922 )
2923 }
2924 Op::Reduce(r) => {
2925 let agent = matches!(r.via, Via::Agent);
2926 (
2927 units_seen,
2928 if agent { node_runs } else { 0 },
2929 0,
2930 if agent {
2931 node_runs as f64 * agent_budget
2932 } else {
2933 0.0
2934 },
2935 )
2936 }
2937 Op::Edit(_) => {
2938 let capped = node_runs.min(w.caps.max_edits);
2939 (
2940 units_seen.min(w.caps.max_edits),
2941 capped,
2942 capped,
2943 capped as f64 * agent_budget,
2944 )
2945 }
2946 Op::Check { .. } | Op::Emit(_) => (units_seen, 0, 0, 0.0),
2947 Op::Call(_) => (units_seen, 0, 0, 0.0),
2950 };
2951
2952 out_units.insert(node.name.as_str(), units_out);
2953 runs = runs.saturating_add(node_runs);
2954 agent_runs = agent_runs.saturating_add(node_agent_runs);
2955 edits = edits.saturating_add(node_edits);
2956 cost += node_cost;
2957 per_node.push(NodeBound {
2958 node: node.name.clone(),
2959 units_in,
2960 units_out,
2961 runs: node_runs,
2962 agent_runs: node_agent_runs,
2963 });
2964 }
2965 Ok(Estimate {
2966 workflow: w.name.clone(),
2967 per_node,
2968 runs,
2969 agent_runs,
2970 edits: edits.min(w.caps.max_edits),
2971 cost,
2972 })
2973 }
2974
2975 fn topological(&self, w: &Workflow) -> Vec<String> {
2978 topological(w)
2979 }
2980}
2981
2982pub fn topological(w: &Workflow) -> Vec<String> {
2985 let forward: Vec<&Edge> = w.edges.iter().filter(|e| e.is_forward()).collect();
2986 let mut indegree: BTreeMap<&str, usize> =
2987 w.nodes.iter().map(|n| (n.name.as_str(), 0)).collect();
2988 for e in &forward {
2989 if let (From::Node(_), To::Node(to)) = (&e.from, &e.to)
2990 && let Some(d) = indegree.get_mut(to.as_str())
2991 {
2992 *d += 1;
2993 }
2994 }
2995 let mut ready: Vec<&str> = w
2996 .nodes
2997 .iter()
2998 .map(|n| n.name.as_str())
2999 .filter(|n| indegree[n] == 0)
3000 .collect();
3001 ready.reverse();
3002 let mut order = Vec::new();
3003 while let Some(n) = ready.pop() {
3004 order.push(n.to_string());
3005 for e in &forward {
3006 if e.from == From::Node(n.to_string())
3007 && let To::Node(to) = &e.to
3008 && let Some(d) = indegree.get_mut(to.as_str())
3009 {
3010 *d -= 1;
3011 if *d == 0 {
3012 ready.insert(0, to.as_str());
3013 }
3014 }
3015 }
3016 }
3017 order
3018}
3019
3020#[cfg(test)]
3021mod tests {
3022 use super::*;
3023
3024 const FIND_ISSUES: &str = r#"{
3026 "types": {
3027 "finding": {"fields": {
3028 "severity": {"type": "enum", "required": true, "values": ["critical", "high"]},
3029 "title": {"type": "line", "required": true, "max": 200, "unique": "normalised"},
3030 "detail": {"type": "lines", "max": 40},
3031 "reason": {"type": "line", "max": 200}
3032 }}
3033 },
3034 "workflow": [
3035 {
3036 "name": "find-issues",
3037 "in": "project", "out": "finding", "effects": [],
3038 "params": {
3039 "review_doc": {"type": "name", "default": "REVIEW.md"},
3040 "breadth": {"type": "int", "default": 20, "max": 40}
3041 },
3042 "caps": {"max_units": 60, "max_edits": 0},
3043 "nodes": [
3044 {"name": "review", "op": "map", "out": "0..n", "in": "project", "emits": "finding",
3045 "max_units": "{$breadth}", "doc": "{$review_doc}",
3046 "task": "Review `{name}`. Write prose to {doc} and findings to {out}."},
3047 {"name": "confirm", "op": "map", "out": "0..1", "in": "finding", "emits": "finding",
3048 "writes": ["reason"],
3049 "task": "Confirm each unit in {in}. Drop what you cannot prove, with a `reason`."},
3050 {"name": "dedupe", "op": "reduce", "via": "rule", "rule": "dedupe",
3051 "group_by": ["title"], "in": "finding", "emits": "finding"}
3052 ],
3053 "edges": [
3054 {"from": "@input", "to": "review"},
3055 {"from": "review", "to": "confirm"},
3056 {"from": "confirm", "to": "dedupe"},
3057 {"from": "dedupe", "to": "@output"}
3058 ]
3059 }
3060 ]
3061 }"#;
3062
3063 fn doc() -> Document {
3064 Document::parse(FIND_ISSUES).expect("the library document parses")
3065 }
3066
3067 #[test]
3069 fn a_parameter_is_renamed_once() {
3070 let rename: BTreeMap<String, String> = [("a", "b"), ("b", "x")]
3071 .into_iter()
3072 .map(|(f, t)| (f.to_string(), t.to_string()))
3073 .collect();
3074 let mut text = "{$a} and {$b}, {$c} and {name} and {$".to_string();
3075 rename_text(&mut text, &rename);
3076 assert_eq!(text, "{$b} and {$x}, {$c} and {name} and {$");
3077 }
3078
3079 #[test]
3081 fn a_callee_parameter_is_not_set_on_the_command_line() {
3082 let d = Document::parse(COMPOSED).unwrap();
3083 let flat = d.flatten("sweep").unwrap();
3084 let e = flat.bind(&["issues/breadth=40".into()]).unwrap_err();
3085 assert!(e.contains("set by the call site `issues`"), "{e}");
3086 let e = flat.bind(&["nope=1".into()]).unwrap_err();
3087 assert!(
3088 !e.contains("issues/"),
3089 "a callee's parameter is not offered: {e}"
3090 );
3091 }
3092
3093 #[test]
3096 fn a_sink_takes_only_the_fields_it_writes() {
3097 let with_sink = |node: &str| {
3098 FIND_ISSUES
3099 .replace(
3100 r#"{"from": "dedupe", "to": "@output"}"#,
3101 r#"{"from": "dedupe", "to": "keep"}, {"from": "keep", "to": "@output"}"#,
3102 )
3103 .replace(
3104 r#""group_by": ["title"], "in": "finding", "emits": "finding"}"#,
3105 &format!(
3106 r#""group_by": ["title"], "in": "finding", "emits": "finding"}}, {node}"#
3107 ),
3108 )
3109 .replace(r#""effects": []"#, r#""effects": ["writes"]"#)
3110 };
3111 let ok = with_sink(
3112 r#"{"name": "keep", "op": "emit", "in": "finding", "sink": "todo", "map": {"text": "{title}", "priority": "high"}}"#,
3113 );
3114 assert!(Document::parse(&ok).is_ok(), "{:?}", Document::parse(&ok));
3115 let typo = with_sink(
3116 r#"{"name": "keep", "op": "emit", "in": "finding", "sink": "todo", "map": {"txet": "{title}", "priority": "high"}}"#,
3117 );
3118 let e = Document::parse(&typo).unwrap_err();
3119 assert!(e.contains("writes no field `txet`"), "{e}");
3120 let unprioritised = with_sink(
3121 r#"{"name": "keep", "op": "emit", "in": "finding", "sink": "todo", "map": {"text": "{title}"}}"#,
3122 );
3123 let e = Document::parse(&unprioritised).unwrap_err();
3124 assert!(e.contains("needs a `priority`"), "{e}");
3125 }
3126
3127 #[test]
3129 fn a_rule_node_emits_the_type_its_rule_writes() {
3130 let wrong = FIND_ISSUES.replace(
3131 r#""max_units": "{$breadth}", "doc": "{$review_doc}",
3132 "task": "Review `{name}`. Write prose to {doc} and findings to {out}."}"#,
3133 r#""max_units": "{$breadth}", "via": "rule", "rule": "todo-items"}"#,
3134 );
3135 let e = Document::parse(&wrong).unwrap_err();
3136 assert!(
3137 e.contains("rule `todo-items` writes `item`, and the node emits `finding`"),
3138 "{e}"
3139 );
3140 }
3141
3142 #[test]
3145 fn unbuilt_constructs_are_named() {
3146 assert!(doc().unbuilt().is_empty(), "{:?}", doc().unbuilt());
3147 let d = Document::parse(WITH_EDIT).unwrap();
3148 let unbuilt = d.unbuilt().join("\n");
3149 assert!(unbuilt.contains("a lap edge is not taken yet"), "{unbuilt}");
3150 assert!(unbuilt.contains("`retry` is not taken yet"), "{unbuilt}");
3151 }
3152
3153 #[test]
3156 fn a_document_round_trips_through_its_stored_form() {
3157 for text in [FIND_ISSUES, WITH_EDIT, COMPOSED] {
3158 let doc = Document::parse(text).unwrap();
3159 let stored = doc.to_json();
3160 assert_eq!(
3161 Document::parse(&stored),
3162 Ok(doc),
3163 "did not round trip:\n{stored}"
3164 );
3165 }
3166 }
3167
3168 #[test]
3169 fn a_library_workflow_parses_with_its_types_and_parameters() {
3170 let d = doc();
3171 let w = d.workflow("find-issues").unwrap();
3172 assert_eq!(w.input, "project");
3173 assert_eq!(w.output.as_deref(), Some("finding"));
3174 assert_eq!(w.effects, Effects::default());
3175 assert_eq!(w.nodes.len(), 3);
3176 assert_eq!(
3177 w.params["breadth"],
3178 Param {
3179 ty: ParamType::Int,
3180 default: Value::from(20),
3181 max: Some(40)
3182 }
3183 );
3184 let Op::Map(review) = &w.nodes[0].op else {
3185 panic!("the first node is a map")
3186 };
3187 assert_eq!(review.out, Out::Grows);
3188 assert_eq!(review.max_units, Some(Count::Param("breadth".into())));
3189 assert_eq!(review.doc.as_deref(), Some("{$review_doc}"));
3190 }
3191
3192 #[test]
3196 fn edges_must_agree_on_the_type_they_carry() {
3197 let broken = FIND_ISSUES.replace(
3198 r#"{"from": "review", "to": "confirm"}"#,
3199 r#"{"from": "@input", "to": "confirm"}"#,
3200 );
3201 let e = Document::parse(&broken).unwrap_err();
3202 assert!(
3203 e.contains("carries `project` into node `confirm`, which reads `finding`"),
3204 "{e}"
3205 );
3206 }
3207
3208 #[test]
3209 fn an_output_edge_must_carry_the_declared_return_type() {
3210 let broken = FIND_ISSUES.replace(
3211 r#""out": "finding", "effects""#,
3212 r#""out": "project", "effects""#,
3213 );
3214 let e = Document::parse(&broken).unwrap_err();
3215 assert!(e.contains("into `@output`, which returns `project`"), "{e}");
3216 }
3217
3218 #[test]
3219 fn a_growing_map_must_bound_what_it_produces() {
3220 let broken = FIND_ISSUES.replace(r#""max_units": "{$breadth}", "#, "");
3221 let e = Document::parse(&broken).unwrap_err();
3222 assert!(e.contains("`0..n` map must state `max_units`"), "{e}");
3223 }
3224
3225 #[test]
3228 fn a_parameter_used_as_a_bound_must_declare_its_maximum() {
3229 let broken = FIND_ISSUES.replace(r#""default": 20, "max": 40"#, r#""default": 20"#);
3230 let e = Document::parse(&broken).unwrap_err();
3231 assert!(e.contains("declares no `max`"), "{e}");
3232 assert!(e.contains("max_units"), "{e}");
3233 }
3234
3235 #[test]
3236 fn every_parameter_needs_a_default() {
3237 let broken = FIND_ISSUES.replace(
3238 r#""type": "name", "default": "REVIEW.md""#,
3239 r#""type": "name""#,
3240 );
3241 let e = Document::parse(&broken).unwrap_err();
3242 assert!(e.contains("has no `default`"), "{e}");
3243 }
3244
3245 #[test]
3249 fn a_declaration_is_checked_where_it_is_written() {
3250 const NAME: &str = r#""review_doc": {"type": "name", "default": "REVIEW.md"}"#;
3251 const INT: &str = r#""breadth": {"type": "int", "default": 20, "max": 40}"#;
3252 for (from, to, want) in [
3253 (
3254 NAME,
3255 r#""review_doc": {"type": "name", "default": 3}"#,
3256 "expected name",
3257 ),
3258 (
3259 NAME,
3260 r#""review_doc": {"type": "name", "default": "../etc/passwd"}"#,
3261 "expected a filename",
3262 ),
3263 (
3264 NAME,
3265 r#""review_doc": {"type": "name", "default": "REVIEW.md", "max": 4}"#,
3266 "`max` bounds an int",
3267 ),
3268 (
3269 INT,
3270 r#""breadth": {"type": "int", "default": true, "max": 40}"#,
3271 "expected int",
3272 ),
3273 (
3274 INT,
3275 r#""breadth": {"type": "int", "default": 50, "max": 40}"#,
3276 "over its own maximum",
3277 ),
3278 (
3279 INT,
3280 r#""breadth": {"type": "int", "default": 20, "max": "forty"}"#,
3281 "`max` must be 0 or more",
3282 ),
3283 ] {
3284 let broken = FIND_ISSUES.replace(from, to);
3285 assert_ne!(broken, FIND_ISSUES, "`{from}` is in the document");
3286 let e = Document::parse(&broken).unwrap_err();
3287 assert!(e.contains(want), "expected `{want}`, got `{e}`");
3288 }
3289 }
3290
3291 #[test]
3292 fn a_field_option_that_does_not_apply_is_refused() {
3293 const REASON: &str = r#""reason": {"type": "line", "max": 200}"#;
3294 for (to, want) in [
3295 (
3296 r#""reason": {"type": "line", "max": "long"}"#,
3297 "`max` must be a whole number",
3298 ),
3299 (
3300 r#""reason": {"type": "line", "max": 200, "unique": true}"#,
3301 "`unique` takes `normalised`",
3302 ),
3303 (
3304 r#""reason": {"type": "line", "max": 200, "min": 2}"#,
3305 "`min` belongs to an int",
3306 ),
3307 (
3308 r#""reason": {"type": "line", "max": 200, "required": "yes"}"#,
3309 "`required` must be true or false",
3310 ),
3311 (
3312 r#""reason": {"type": "line", "max": 200, "values": ["a"]}"#,
3313 "`values` belongs to an enum",
3314 ),
3315 (
3316 r#""reason": {"type": "line", "max": 0}"#,
3317 "`max` must be 1 or more",
3318 ),
3319 ] {
3320 let broken = FIND_ISSUES.replace(REASON, to);
3321 assert_ne!(broken, FIND_ISSUES, "the reason field is in the document");
3322 let e = Document::parse(&broken).unwrap_err();
3323 assert!(e.contains(want), "expected `{want}`, got `{e}`");
3324 }
3325 }
3326
3327 #[test]
3328 fn placeholders_are_checked_against_what_they_could_read() {
3329 for (from, to, want) in [
3330 ("{$review_doc}", "{$missing}", "names no parameter"),
3331 ("Review `{name}`", "Review `{title}`", "names no field"),
3332 ] {
3333 let broken = FIND_ISSUES.replace(from, to);
3334 let e = Document::parse(&broken).unwrap_err();
3335 assert!(e.contains(want), "{from} -> {to}: {e}");
3336 }
3337 }
3338
3339 #[test]
3340 fn a_type_may_not_shadow_a_built_in_or_a_system_field() {
3341 let shadow = FIND_ISSUES.replace(r#""finding": {"fields""#, r#""project": {"fields""#);
3342 assert!(
3343 Document::parse(&shadow)
3344 .unwrap_err()
3345 .contains("built-in type")
3346 );
3347 let reserved = FIND_ISSUES.replace(
3348 r#""reason": {"type": "line""#,
3349 r#""@id": {"type": "line""#,
3350 );
3351 assert!(Document::parse(&reserved).unwrap_err().contains("reserved"));
3352 }
3353
3354 #[test]
3355 fn an_unreachable_node_is_refused() {
3356 let broken = FIND_ISSUES.replace(r#"{"from": "confirm", "to": "dedupe"},"#, "");
3357 let e = Document::parse(&broken).unwrap_err();
3358 assert!(e.contains("node `dedupe` is not reachable"), "{e}");
3359 }
3360
3361 #[test]
3362 fn an_unknown_field_is_refused_rather_than_ignored() {
3363 let broken = FIND_ISSUES.replace(
3364 r#""name": "find-issues","#,
3365 r#""name": "find-issues", "retries": 2,"#,
3366 );
3367 let e = Document::parse(&broken).unwrap_err();
3368 assert!(e.contains("unknown field `retries`"), "{e}");
3369 }
3370
3371 #[test]
3374 fn the_bound_is_computed_at_the_declared_maximum() {
3375 let d = doc();
3376 let e = d.estimate("find-issues", 1, 1.0).unwrap();
3377 assert_eq!(e.per_node[0].units_out, 40, "review at breadth's max");
3378 assert_eq!(e.per_node[1].units_in, 40, "confirm sees them all");
3379 assert_eq!(e.agent_runs, 41, "one review, forty confirms");
3380 assert_eq!(e.edits, 0);
3381 assert!((e.cost - 41.0).abs() < f64::EPSILON, "{}", e.cost);
3382 }
3383
3384 #[test]
3386 fn rule_nodes_are_free() {
3387 let d = doc();
3388 let e = d.estimate("find-issues", 1, 1.0).unwrap();
3389 let dedupe = e.per_node.iter().find(|b| b.node == "dedupe").unwrap();
3390 assert_eq!(dedupe.agent_runs, 0);
3391 assert_eq!(dedupe.runs, 40, "it still handles every unit");
3392 }
3393
3394 const WITH_EDIT: &str = r#"{
3395 "types": {"finding": {"fields": {
3396 "title": {"type": "line", "required": true, "max": 200},
3397 "verdict": {"type": "enum", "values": ["accept", "reject"]},
3398 "reason": {"type": "line", "max": 200}
3399 }}},
3400 "workflow": [{
3401 "name": "apply-fixes",
3402 "in": "finding", "out": "finding", "effects": ["repo", "writes"],
3403 "params": {"laps": {"type": "int", "default": 2, "max": 3},
3404 "retries": {"type": "int", "default": 1, "max": 2}},
3405 "caps": {"max_units": 40, "max_edits": 12},
3406 "nodes": [
3407 {"name": "fix", "op": "edit", "in": "finding", "check": "verify",
3408 "retry": {"max": "{$retries}", "while": "@verify != passed"},
3409 "task": "Fix `{title}` in `{@project}`."},
3410 {"name": "audit", "op": "map", "out": "1", "in": "finding", "emits": "finding",
3411 "writes": ["verdict", "reason"],
3412 "task": "Judge the diff for `{title}`. Set `verdict` and `reason`."},
3413 {"name": "handoff", "op": "emit", "in": "finding", "sink": "todo", "action": "add",
3414 "map": {"text": "{title}", "priority": "high", "description": "{reason}"}}
3415 ],
3416 "edges": [
3417 {"from": "@input", "to": "fix"},
3418 {"from": "fix", "to": "audit"},
3419 {"from": "audit", "to": "@output", "when": {"verdict": ["accept"]}},
3420 {"from": "audit", "to": "fix", "when": {"verdict": ["reject"]}, "max_laps": "{$laps}"},
3421 {"from": "audit", "to": "handoff", "default": true}
3422 ]
3423 }]
3424 }"#;
3425
3426 #[test]
3427 fn an_edit_and_a_sink_are_read_back_as_effects() {
3428 let d = Document::parse(WITH_EDIT).unwrap();
3429 let w = d.workflow("apply-fixes").unwrap();
3430 assert_eq!(
3431 w.effects,
3432 Effects {
3433 repo: true,
3434 writes: true
3435 }
3436 );
3437 }
3438
3439 #[test]
3442 fn declared_effects_must_match_the_graph() {
3443 let lying = WITH_EDIT.replace(r#""effects": ["repo", "writes"]"#, r#""effects": []"#);
3444 let e = Document::parse(&lying).unwrap_err();
3445 assert!(e.contains("declares effects pure"), "{e}");
3446 assert!(e.contains("repo, writes"), "{e}");
3447 }
3448
3449 #[test]
3452 fn a_lap_edge_needs_a_terminal_path() {
3453 let broken = WITH_EDIT.replace(
3454 r#",
3455 {"from": "audit", "to": "handoff", "default": true}"#,
3456 "",
3457 );
3458 let e = Document::parse(&broken).unwrap_err();
3459 assert!(e.contains("lap edge and no default edge"), "{e}");
3460 }
3461
3462 #[test]
3463 fn a_cycle_that_is_not_a_lap_edge_is_refused() {
3464 let broken = WITH_EDIT.replace(r#", "max_laps": "{$laps}""#, "");
3465 let e = Document::parse(&broken).unwrap_err();
3466 assert!(
3467 e.contains("cycle that is not a `max_laps` back edge"),
3468 "{e}"
3469 );
3470 }
3471
3472 #[test]
3475 fn laps_and_retries_multiply_and_the_cap_truncates() {
3476 let d = Document::parse(WITH_EDIT).unwrap();
3477 let e = d.estimate("apply-fixes", 2, 1.0).unwrap();
3478 let fix = e.per_node.iter().find(|b| b.node == "fix").unwrap();
3479 assert_eq!(fix.runs, 24);
3481 assert_eq!(fix.agent_runs, 12, "max_edits truncates");
3482 assert_eq!(e.edits, 12);
3483 }
3484
3485 #[test]
3488 fn a_lap_mints_a_unit_but_a_retry_does_not() {
3489 let d = Document::parse(WITH_EDIT).unwrap();
3490 let e = d.estimate("apply-fixes", 1, 1.0).unwrap();
3491 let at = |name: &str| e.per_node.iter().find(|b| b.node == name).unwrap();
3492 assert_eq!(at("fix").units_out, 4);
3494 assert_eq!(at("fix").runs, 12);
3495 assert_eq!(at("audit").units_in, 4);
3497 assert_eq!(at("audit").runs, 4);
3498 }
3499
3500 #[test]
3501 fn a_guard_may_only_read_what_its_units_carry() {
3502 let broken = WITH_EDIT.replace(r#"{"verdict": ["accept"]}"#, r#"{"outcome": ["accept"]}"#);
3503 let e = Document::parse(&broken).unwrap_err();
3504 assert!(e.contains("`outcome` is no field of `finding`"), "{e}");
3505 let bad_value =
3506 WITH_EDIT.replace(r#"{"verdict": ["accept"]}"#, r#"{"verdict": ["maybe"]}"#);
3507 let e = Document::parse(&bad_value).unwrap_err();
3508 assert!(e.contains("not one of that enum's values"), "{e}");
3509 }
3510
3511 #[test]
3512 fn a_verdict_guard_names_a_check_the_graph_runs() {
3513 assert!(Document::parse(WITH_EDIT).is_ok());
3515 let broken = WITH_EDIT.replace(
3516 r#"{"from": "fix", "to": "audit"}"#,
3517 r#"{"from": "fix", "to": "audit", "when": {"@merged": ["passed"]}}"#,
3518 );
3519 let e = Document::parse(&broken).unwrap_err();
3520 assert!(
3521 e.contains("`@merged` names no system field or verdict"),
3522 "{e}"
3523 );
3524 }
3525
3526 const COMPOSED: &str = r#"{
3527 "types": {"finding": {"fields": {
3528 "severity": {"type": "enum", "required": true, "values": ["critical", "high"]},
3529 "title": {"type": "line", "required": true, "max": 200}
3530 }}},
3531 "workflow": [
3532 {"name": "look", "in": "project", "out": "finding", "effects": [],
3533 "params": {"breadth": {"type": "int", "default": 5, "max": 10}},
3534 "caps": {"max_units": 40, "max_edits": 0},
3535 "nodes": [{"name": "review", "op": "map", "out": "0..n", "in": "project",
3536 "emits": "finding", "max_units": "{$breadth}",
3537 "task": "Review `{name}`."}],
3538 "edges": [{"from": "@input", "to": "review"},
3539 {"from": "review", "to": "@output"}]},
3540 {"name": "sweep", "in": "project", "out": "finding", "effects": ["writes"],
3541 "params": {},
3542 "caps": {"max_units": 80, "max_edits": 0},
3543 "nodes": [
3544 {"name": "issues", "op": "call", "workflow": "look", "in": "project",
3545 "with": {"breadth": 4}},
3546 {"name": "keep", "op": "emit", "in": "finding", "sink": "todo", "action": "add",
3547 "map": {"text": "{title}", "priority": "{severity}"}}
3548 ],
3549 "edges": [{"from": "@input", "to": "issues"},
3550 {"from": "issues", "to": "keep"},
3551 {"from": "keep", "to": "@output"}]}
3552 ]
3553 }"#;
3554
3555 #[test]
3556 fn a_call_takes_its_callees_signature() {
3557 let d = Document::parse(COMPOSED).unwrap();
3558 let sweep = d.workflow("sweep").unwrap();
3559 assert_eq!(
3560 sweep.effects,
3561 Effects {
3562 repo: false,
3563 writes: true
3564 }
3565 );
3566 let e = d.estimate("sweep", 3, 1.0).unwrap();
3570 let issues = e
3571 .per_node
3572 .iter()
3573 .find(|b| b.node == "issues/review")
3574 .unwrap();
3575 assert_eq!(issues.units_out, 12);
3576 assert_eq!(e.agent_runs, 3, "one review run per project");
3577 }
3578
3579 #[test]
3583 fn a_call_is_replaced_by_its_callee() {
3584 let d = Document::parse(COMPOSED).unwrap();
3585 let flat = d.flatten("sweep").unwrap();
3586 assert_eq!(
3587 flat.nodes
3588 .iter()
3589 .map(|n| n.name.as_str())
3590 .collect::<Vec<_>>(),
3591 ["keep", "issues/review"]
3592 );
3593 let breadth = flat
3594 .params
3595 .get("issues/breadth")
3596 .expect("the call's argument");
3597 assert_eq!(breadth.default, Value::from(4), "`with` set the default");
3598 assert_eq!(
3599 breadth.max,
3600 Some(4),
3601 "a pinned constant is the bound, since nothing can change it"
3602 );
3603 let Op::Map(m) = &flat.node("issues/review").unwrap().op else {
3604 panic!("the callee's node")
3605 };
3606 assert_eq!(m.max_units, Some(Count::Param("issues/breadth".into())));
3607
3608 assert!(
3610 flat.edges
3611 .iter()
3612 .any(|e| e.from == From::Input && e.to == To::Node("issues/review".into()))
3613 );
3614 assert!(flat.edges.iter().any(
3615 |e| e.from == From::Node("issues/review".into()) && e.to == To::Node("keep".into())
3616 ));
3617 assert!(!flat.edges.iter().any(|e| e.to == To::Node("issues".into())));
3618 }
3619
3620 #[test]
3621 fn a_call_must_agree_with_what_its_callee_reads() {
3622 let broken = COMPOSED.replace(
3623 r#""workflow": "look", "in": "project""#,
3624 r#""workflow": "look", "in": "finding""#,
3625 );
3626 let e = Document::parse(&broken).unwrap_err();
3627 assert!(e.contains("reads `project`"), "{e}");
3628 }
3629
3630 #[test]
3631 fn a_call_may_not_pass_a_value_over_the_callees_maximum() {
3632 let broken = COMPOSED.replace(r#""with": {"breadth": 4}"#, r#""with": {"breadth": 11}"#);
3633 let e = Document::parse(&broken).unwrap_err();
3634 assert!(e.contains("`breadth` is 11, over the maximum of 10"), "{e}");
3635 }
3636
3637 #[test]
3638 fn a_call_may_not_pass_an_undeclared_parameter() {
3639 let broken = COMPOSED.replace(r#""with": {"breadth": 4}"#, r#""with": {"depth": 4}"#);
3640 let e = Document::parse(&broken).unwrap_err();
3641 assert!(e.contains("declares no parameter `depth`"), "{e}");
3642 }
3643
3644 #[test]
3647 fn a_workflow_may_not_call_itself() {
3648 let broken = COMPOSED.replace(
3649 r#""workflow": "look", "in": "project",
3650 "with": {"breadth": 4}"#,
3651 r#""workflow": "sweep", "in": "project""#,
3652 );
3653 let e = Document::parse(&broken).unwrap_err();
3654 assert!(e.contains("calls itself"), "{e}");
3655 }
3656}