1use serde::{Deserialize, Serialize};
14use serde_json::Value;
15use sha2::{Digest, Sha256};
16use std::collections::{BTreeMap, BTreeSet};
17
18const PROPOSAL_KIND: &str = "workflow_proposal";
19const PROPOSAL_SCHEMA_VERSION: &str = "1.0.0";
20const PROPOSAL_DIGEST_VERSION: &str = "1.0.0";
21
22#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
26pub struct WorkflowProposal {
27 pub kind: String,
28 pub schema_version: String,
29 pub proposal_id: String,
30 pub workspace_id: String,
32 pub app_manifest: ManifestReference,
33 pub nodes: Vec<ProposalNode>,
34 pub edges: Vec<ProposalEdge>,
35 pub mappings: Vec<ProposalMapping>,
36 pub initial_input: Value,
39}
40
41#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
45pub struct ManifestReference {
46 pub app_id: String,
47 pub app_version: String,
48 pub manifest_digest: String,
49}
50
51#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
54pub struct ProposalNode {
55 pub node_id: String,
56 pub capability_id: String,
57 pub capability_version: String,
58 pub artifact_digest: String,
59}
60
61#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
64pub struct ProposalEdge {
65 pub from_node_id: String,
66 pub to_node_id: String,
67}
68
69#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
73pub struct ProposalMapping {
74 pub source: MappingSource,
75 pub source_path: String,
77 pub target_node_id: String,
78 pub target_path: String,
80}
81
82#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
83#[serde(rename_all = "snake_case", tag = "kind")]
84pub enum MappingSource {
85 InitialInput,
86 Node { node_id: String },
87}
88
89#[derive(Debug, Clone, Copy, PartialEq, Eq)]
92pub struct ProposalLimits {
93 pub max_nodes: usize,
94 pub max_edges: usize,
95 pub max_mappings: usize,
96 pub max_initial_input_bytes: usize,
97}
98
99pub const DEFAULT_MAX_PROPOSAL_NODES: usize = 32;
100pub const DEFAULT_MAX_PROPOSAL_EDGES: usize = 64;
101pub const DEFAULT_MAX_PROPOSAL_MAPPINGS: usize = 128;
102pub const DEFAULT_MAX_INITIAL_INPUT_BYTES: usize = 262_144;
103
104impl Default for ProposalLimits {
105 fn default() -> Self {
106 Self {
107 max_nodes: DEFAULT_MAX_PROPOSAL_NODES,
108 max_edges: DEFAULT_MAX_PROPOSAL_EDGES,
109 max_mappings: DEFAULT_MAX_PROPOSAL_MAPPINGS,
110 max_initial_input_bytes: DEFAULT_MAX_INITIAL_INPUT_BYTES,
111 }
112 }
113}
114
115#[derive(Debug, Clone, PartialEq)]
118pub struct CanonicalProposal {
119 pub proposal: WorkflowProposal,
120 pub execution_order: Vec<String>,
122}
123
124#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
125pub struct ProposalValidationFailure {
126 pub errors: Vec<ProposalValidationError>,
127}
128
129#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
130pub struct ProposalValidationError {
131 pub code: ProposalValidationErrorCode,
132 pub message: String,
133 pub path: String,
134}
135
136#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
137#[serde(rename_all = "snake_case")]
138pub enum ProposalValidationErrorCode {
139 InvalidLiteral,
140 MissingRequiredField,
141 NodeLimitExceeded,
142 EdgeLimitExceeded,
143 MappingLimitExceeded,
144 PayloadLimitExceeded,
145 DuplicateNodeId,
146 UnknownEdgeEndpoint,
147 SelfLoopEdge,
148 DuplicateEdge,
149 CyclicGraph,
150 UnknownMappingEndpoint,
151 AmbiguousMultiWriterTarget,
152 MissingDependencyEdgeForMapping,
153}
154
155#[allow(clippy::too_many_lines)]
164pub fn canonicalize_proposal(
165 proposal: WorkflowProposal,
166 limits: &ProposalLimits,
167) -> Result<CanonicalProposal, ProposalValidationFailure> {
168 let mut errors = Vec::new();
169
170 if proposal.kind != PROPOSAL_KIND {
171 errors.push(error(
172 ProposalValidationErrorCode::InvalidLiteral,
173 "$.kind",
174 "kind must equal workflow_proposal",
175 ));
176 }
177 if proposal.schema_version != PROPOSAL_SCHEMA_VERSION {
178 errors.push(error(
179 ProposalValidationErrorCode::InvalidLiteral,
180 "$.schema_version",
181 "schema_version must equal 1.0.0",
182 ));
183 }
184 validate_non_empty(&proposal.proposal_id, "$.proposal_id", &mut errors);
185 validate_non_empty(&proposal.workspace_id, "$.workspace_id", &mut errors);
186
187 if proposal.nodes.len() > limits.max_nodes {
188 errors.push(error(
189 ProposalValidationErrorCode::NodeLimitExceeded,
190 "$.nodes",
191 &format!(
192 "proposal declares {} nodes, exceeding the configured limit of {}",
193 proposal.nodes.len(),
194 limits.max_nodes
195 ),
196 ));
197 }
198 if proposal.edges.len() > limits.max_edges {
199 errors.push(error(
200 ProposalValidationErrorCode::EdgeLimitExceeded,
201 "$.edges",
202 &format!(
203 "proposal declares {} edges, exceeding the configured limit of {}",
204 proposal.edges.len(),
205 limits.max_edges
206 ),
207 ));
208 }
209 if proposal.mappings.len() > limits.max_mappings {
210 errors.push(error(
211 ProposalValidationErrorCode::MappingLimitExceeded,
212 "$.mappings",
213 &format!(
214 "proposal declares {} mappings, exceeding the configured limit of {}",
215 proposal.mappings.len(),
216 limits.max_mappings
217 ),
218 ));
219 }
220 let initial_input_bytes =
221 serde_json::to_vec(&proposal.initial_input).map_or(usize::MAX, |bytes| bytes.len());
222 if initial_input_bytes > limits.max_initial_input_bytes {
223 errors.push(error(
224 ProposalValidationErrorCode::PayloadLimitExceeded,
225 "$.initial_input",
226 &format!(
227 "initial_input is {initial_input_bytes} bytes, exceeding the configured limit of \
228 {} bytes",
229 limits.max_initial_input_bytes
230 ),
231 ));
232 }
233
234 let mut node_ids: BTreeSet<String> = BTreeSet::new();
235 for (index, node) in proposal.nodes.iter().enumerate() {
236 let path = format!("$.nodes[{index}].node_id");
237 validate_non_empty(&node.node_id, &path, &mut errors);
238 validate_non_empty(
239 &node.capability_id,
240 &format!("$.nodes[{index}].capability_id"),
241 &mut errors,
242 );
243 validate_non_empty(
244 &node.capability_version,
245 &format!("$.nodes[{index}].capability_version"),
246 &mut errors,
247 );
248 validate_non_empty(
249 &node.artifact_digest,
250 &format!("$.nodes[{index}].artifact_digest"),
251 &mut errors,
252 );
253 if !node_ids.insert(node.node_id.clone()) {
254 errors.push(error(
255 ProposalValidationErrorCode::DuplicateNodeId,
256 &path,
257 &format!("node_id '{}' is declared more than once", node.node_id),
258 ));
259 }
260 }
261
262 let mut adjacency: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
263 let mut in_degree: BTreeMap<String, usize> =
264 node_ids.iter().map(|id| (id.clone(), 0)).collect();
265 let mut declared_edges: BTreeSet<(String, String)> = BTreeSet::new();
266 for (index, edge) in proposal.edges.iter().enumerate() {
267 let path = format!("$.edges[{index}]");
268 if edge.from_node_id == edge.to_node_id {
269 errors.push(error(
270 ProposalValidationErrorCode::SelfLoopEdge,
271 &path,
272 &format!("edge from '{}' to itself is not allowed", edge.from_node_id),
273 ));
274 continue;
275 }
276 if !node_ids.contains(&edge.from_node_id) {
277 errors.push(error(
278 ProposalValidationErrorCode::UnknownEdgeEndpoint,
279 &format!("{path}.from_node_id"),
280 &format!("edge references unknown node_id '{}'", edge.from_node_id),
281 ));
282 continue;
283 }
284 if !node_ids.contains(&edge.to_node_id) {
285 errors.push(error(
286 ProposalValidationErrorCode::UnknownEdgeEndpoint,
287 &format!("{path}.to_node_id"),
288 &format!("edge references unknown node_id '{}'", edge.to_node_id),
289 ));
290 continue;
291 }
292 let key = (edge.from_node_id.clone(), edge.to_node_id.clone());
293 if !declared_edges.insert(key) {
294 errors.push(error(
295 ProposalValidationErrorCode::DuplicateEdge,
296 &path,
297 &format!(
298 "edge '{}' -> '{}' is declared more than once",
299 edge.from_node_id, edge.to_node_id
300 ),
301 ));
302 continue;
303 }
304 adjacency
305 .entry(edge.from_node_id.clone())
306 .or_default()
307 .insert(edge.to_node_id.clone());
308 *in_degree.entry(edge.to_node_id.clone()).or_insert(0) += 1;
309 }
310
311 let mut writer_targets: BTreeMap<(String, String), usize> = BTreeMap::new();
312 for (index, mapping) in proposal.mappings.iter().enumerate() {
313 let path = format!("$.mappings[{index}]");
314 validate_non_empty(
315 &mapping.source_path,
316 &format!("{path}.source_path"),
317 &mut errors,
318 );
319 validate_non_empty(
320 &mapping.target_path,
321 &format!("{path}.target_path"),
322 &mut errors,
323 );
324 if !node_ids.contains(&mapping.target_node_id) {
325 errors.push(error(
326 ProposalValidationErrorCode::UnknownMappingEndpoint,
327 &format!("{path}.target_node_id"),
328 &format!(
329 "mapping targets unknown node_id '{}'",
330 mapping.target_node_id
331 ),
332 ));
333 continue;
334 }
335 if let MappingSource::Node { node_id } = &mapping.source {
336 if !node_ids.contains(node_id) {
337 errors.push(error(
338 ProposalValidationErrorCode::UnknownMappingEndpoint,
339 &format!("{path}.source"),
340 &format!("mapping sources unknown node_id '{node_id}'"),
341 ));
342 continue;
343 }
344 if !declared_edges.contains(&(node_id.clone(), mapping.target_node_id.clone())) {
345 errors.push(error(
346 ProposalValidationErrorCode::MissingDependencyEdgeForMapping,
347 &path,
348 &format!(
349 "mapping from '{node_id}' to '{}' has no corresponding declared edge",
350 mapping.target_node_id
351 ),
352 ));
353 continue;
354 }
355 }
356 let writer_key = (mapping.target_node_id.clone(), mapping.target_path.clone());
357 *writer_targets.entry(writer_key).or_insert(0) += 1;
358 }
359 for ((target_node_id, target_path), count) in &writer_targets {
360 if *count > 1 {
361 errors.push(error(
362 ProposalValidationErrorCode::AmbiguousMultiWriterTarget,
363 &format!(
364 "$.mappings[?target_node_id={target_node_id}][?target_path={target_path}]"
365 ),
366 &format!(
367 "target path '{target_path}' on node '{target_node_id}' is written by {count} \
368 mappings; a target path may have at most one writer"
369 ),
370 ));
371 }
372 }
373
374 if !errors.is_empty() {
375 return Err(ProposalValidationFailure { errors });
376 }
377
378 let Ok(execution_order) = topological_order(&node_ids, &adjacency, &in_degree) else {
379 return Err(ProposalValidationFailure {
380 errors: vec![error(
381 ProposalValidationErrorCode::CyclicGraph,
382 "$.edges",
383 "proposal graph contains a cycle; P1 requires an acyclic graph",
384 )],
385 });
386 };
387
388 Ok(CanonicalProposal {
389 proposal,
390 execution_order,
391 })
392}
393
394fn topological_order(
397 node_ids: &BTreeSet<String>,
398 adjacency: &BTreeMap<String, BTreeSet<String>>,
399 in_degree: &BTreeMap<String, usize>,
400) -> Result<Vec<String>, ()> {
401 let mut remaining_in_degree = in_degree.clone();
402 let mut ready: BTreeSet<String> = node_ids
403 .iter()
404 .filter(|id| remaining_in_degree.get(*id).copied().unwrap_or(0) == 0)
405 .cloned()
406 .collect();
407 let mut order = Vec::with_capacity(node_ids.len());
408
409 while let Some(next) = ready.iter().next().cloned() {
410 ready.remove(&next);
411 order.push(next.clone());
412 let Some(successors) = adjacency.get(&next) else {
413 continue;
414 };
415 for successor in successors {
416 let degree = remaining_in_degree.entry(successor.clone()).or_insert(0);
420 *degree -= 1;
421 if *degree == 0 {
422 ready.insert(successor.clone());
423 }
424 }
425 }
426
427 if order.len() == node_ids.len() {
428 Ok(order)
429 } else {
430 Err(())
431 }
432}
433
434#[derive(Debug, Clone, Copy, PartialEq, Eq)]
442pub struct ParallelScheduleLimits {
443 pub max_fan_out: usize,
446 pub max_join_width: usize,
448 pub max_queue_depth: usize,
450 pub max_concurrent_nodes: usize,
452}
453
454pub const DEFAULT_MAX_FAN_OUT: usize = 8;
455pub const DEFAULT_MAX_JOIN_WIDTH: usize = 8;
456pub const DEFAULT_MAX_QUEUE_DEPTH: usize = 16;
457pub const DEFAULT_MAX_CONCURRENT_NODES: usize = 8;
458
459impl Default for ParallelScheduleLimits {
460 fn default() -> Self {
461 Self {
462 max_fan_out: DEFAULT_MAX_FAN_OUT,
463 max_join_width: DEFAULT_MAX_JOIN_WIDTH,
464 max_queue_depth: DEFAULT_MAX_QUEUE_DEPTH,
465 max_concurrent_nodes: DEFAULT_MAX_CONCURRENT_NODES,
466 }
467 }
468}
469
470#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
471pub struct ParallelScheduleError {
472 pub code: ParallelScheduleErrorCode,
473 pub message: String,
474 pub path: String,
475}
476
477#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize)]
478#[serde(rename_all = "snake_case")]
479pub enum ParallelScheduleErrorCode {
480 FanOutExceeded,
481 JoinWidthExceeded,
482 QueueDepthExceeded,
483}
484
485#[derive(Debug, Clone, PartialEq, Eq)]
486pub struct ParallelScheduleFailure {
487 pub errors: Vec<ParallelScheduleError>,
488}
489
490#[derive(Debug, Clone, PartialEq, Eq)]
495pub struct ParallelSchedule {
496 pub waves: Vec<Vec<String>>,
497}
498
499pub fn compute_parallel_schedule(
510 canonical: &CanonicalProposal,
511 limits: &ParallelScheduleLimits,
512) -> Result<ParallelSchedule, ParallelScheduleFailure> {
513 let mut errors = Vec::new();
514
515 let node_ids: BTreeSet<String> = canonical
516 .proposal
517 .nodes
518 .iter()
519 .map(|node| node.node_id.clone())
520 .collect();
521 if node_ids.len() > limits.max_queue_depth {
522 errors.push(ParallelScheduleError {
523 code: ParallelScheduleErrorCode::QueueDepthExceeded,
524 message: format!(
525 "schedule has {} nodes, exceeding the configured queue depth of {}",
526 node_ids.len(),
527 limits.max_queue_depth
528 ),
529 path: "$.nodes".to_string(),
530 });
531 }
532
533 let mut adjacency: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
534 let mut in_degree: BTreeMap<String, usize> =
535 node_ids.iter().map(|id| (id.clone(), 0)).collect();
536 for edge in &canonical.proposal.edges {
537 adjacency
538 .entry(edge.from_node_id.clone())
539 .or_default()
540 .insert(edge.to_node_id.clone());
541 *in_degree.entry(edge.to_node_id.clone()).or_insert(0) += 1;
542 }
543
544 for (node_id, degree) in &in_degree {
545 if *degree > limits.max_join_width {
546 errors.push(ParallelScheduleError {
547 code: ParallelScheduleErrorCode::JoinWidthExceeded,
548 message: format!(
549 "node '{node_id}' has {degree} direct predecessors, exceeding the configured join width of {}",
550 limits.max_join_width
551 ),
552 path: format!("$.nodes[?node_id={node_id}]"),
553 });
554 }
555 }
556
557 let mut remaining_in_degree = in_degree.clone();
558 let mut ready: BTreeSet<String> = node_ids
559 .iter()
560 .filter(|id| remaining_in_degree.get(*id).copied().unwrap_or(0) == 0)
561 .cloned()
562 .collect();
563 let mut waves = Vec::new();
564 while !ready.is_empty() {
565 let wave: Vec<String> = ready.iter().cloned().collect();
566 if wave.len() > limits.max_fan_out {
567 errors.push(ParallelScheduleError {
568 code: ParallelScheduleErrorCode::FanOutExceeded,
569 message: format!(
570 "{} nodes became ready concurrently, exceeding the configured fan-out of {}",
571 wave.len(),
572 limits.max_fan_out
573 ),
574 path: "$.nodes".to_string(),
575 });
576 }
577 let mut next_ready = BTreeSet::new();
578 for node_id in &wave {
579 if let Some(successors) = adjacency.get(node_id) {
580 for successor in successors {
581 let degree = remaining_in_degree.entry(successor.clone()).or_insert(0);
582 *degree -= 1;
583 if *degree == 0 {
584 next_ready.insert(successor.clone());
585 }
586 }
587 }
588 }
589 waves.push(wave);
590 ready = next_ready;
591 }
592
593 if errors.is_empty() {
594 Ok(ParallelSchedule { waves })
595 } else {
596 Err(ParallelScheduleFailure { errors })
597 }
598}
599
600#[must_use]
605pub fn proposal_digest(proposal: &WorkflowProposal) -> String {
606 let value = serde_json::to_value(proposal).unwrap_or(Value::Null);
607 digest_json_value(&value)
608}
609
610#[must_use]
616pub fn proposal_snapshot_digest(proposal_digest: &str, snapshots: &SnapshotDigests) -> String {
617 let value = serde_json::json!({
618 "proposal_digest": proposal_digest,
619 "manifest_digest": snapshots.manifest_digest,
620 "registry_digest": snapshots.registry_digest,
621 "binding_digest": snapshots.binding_digest,
622 "policy_digest": snapshots.policy_digest,
623 "budget_digest": snapshots.budget_digest,
624 });
625 digest_json_value(&value)
626}
627
628#[derive(Debug, Clone, PartialEq, Eq)]
633pub struct SnapshotDigests {
634 pub manifest_digest: String,
635 pub registry_digest: String,
636 pub binding_digest: String,
637 pub policy_digest: String,
638 pub budget_digest: String,
639}
640
641fn digest_json_value(value: &Value) -> String {
642 let canonical = canonical_json_string(value);
643 let mut hasher = Sha256::new();
644 hasher.update(canonical.as_bytes());
645 let digest = hasher.finalize();
646 format!("{PROPOSAL_DIGEST_VERSION}:sha256:{}", hex_encode(&digest))
647}
648
649fn hex_encode(bytes: &[u8]) -> String {
650 use std::fmt::Write as _;
651 let mut out = String::with_capacity(bytes.len() * 2);
652 for byte in bytes {
653 let _ = write!(out, "{byte:02x}");
654 }
655 out
656}
657
658fn canonical_json_string(value: &Value) -> String {
662 let mut out = String::new();
663 write_canonical(value, &mut out);
664 out
665}
666
667fn write_canonical(value: &Value, out: &mut String) {
668 match value {
669 Value::Null | Value::Bool(_) | Value::Number(_) => {
670 out.push_str(&value.to_string());
671 }
672 Value::String(s) => {
673 out.push_str(&serde_json::to_string(s).unwrap_or_default());
674 }
675 Value::Array(items) => {
676 out.push('[');
677 for (index, item) in items.iter().enumerate() {
678 if index > 0 {
679 out.push(',');
680 }
681 write_canonical(item, out);
682 }
683 out.push(']');
684 }
685 Value::Object(map) => {
686 out.push('{');
687 let mut keys: Vec<&String> = map.keys().collect();
688 keys.sort();
689 for (index, key) in keys.iter().enumerate() {
690 if index > 0 {
691 out.push(',');
692 }
693 out.push_str(&serde_json::to_string(key).unwrap_or_default());
694 out.push(':');
695 write_canonical(&map[*key], out);
696 }
697 out.push('}');
698 }
699 }
700}
701
702fn validate_non_empty(value: &str, path: &str, errors: &mut Vec<ProposalValidationError>) {
703 if value.trim().is_empty() {
704 errors.push(error(
705 ProposalValidationErrorCode::MissingRequiredField,
706 path,
707 "value must be non-empty",
708 ));
709 }
710}
711
712fn error(code: ProposalValidationErrorCode, path: &str, message: &str) -> ProposalValidationError {
713 ProposalValidationError {
714 code,
715 message: message.to_string(),
716 path: path.to_string(),
717 }
718}