1mod elevation;
8mod gates;
9mod js_authoring;
10mod model_policy;
11mod named_fleet;
12mod replay;
13mod role_resolve;
14
15use std::collections::{BTreeMap, BTreeSet};
16use std::path::Path;
17
18use serde::{Deserialize, Serialize};
19use thiserror::Error;
20
21pub use elevation::{
22 DEFAULT_HIGH_BUDGET_THRESHOLD, ElevationOptions, PlanRiskHint, WorkflowPlanElevation,
23 assess_plan_risk_string, assess_workflow_elevation,
24};
25pub use gates::{
26 GateError, GateKind, GateOn, GateOnFail, GateOutcome, GateSpec, GateState, GateStatusLine,
27 HandoffArtifact, LaneGateBoard, stopship_gate_pipeline,
28};
29pub use js_authoring::{
30 JavascriptWorkflowError, JavascriptWorkflowResult, compile_javascript_workflow,
31 compile_typescript_workflow,
32};
33pub use model_policy::*;
34pub use named_fleet::{
35 NamedFleet, NamedFleetError, STOPSHIP_REQUIRED_ROLES, load_named_fleet, load_named_fleet_file,
36 parse_named_fleet,
37};
38pub use replay::*;
39pub use role_resolve::{
40 FleetRoleMap, FleetRoleResolveError, ResolvedWorkflowAgent, normalize_token,
41 resolve_workflow_agent, validate_role_token,
42};
43
44pub const DEFAULT_FLEET_WORKFLOW_MAX_AGENTS: usize = 1000;
47pub const DEFAULT_FLEET_WORKFLOW_MAX_DEPTH: usize = 5;
48
49#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
50pub struct WorkflowConfig {
51 pub goal: String,
52 #[serde(default = "default_max_concurrent")]
53 pub max_concurrent: u8,
54 #[serde(default)]
55 pub description: Option<String>,
56 #[serde(default)]
57 pub phases: Vec<Phase>,
58}
59
60impl WorkflowConfig {
61 pub fn validate(&self) -> Result<(), WorkflowValidationError> {
62 WorkflowPlan::from_config(self).map(|_| ())
63 }
64
65 pub fn compile(&self) -> Result<WorkflowPlan, WorkflowValidationError> {
66 WorkflowPlan::from_config(self)
67 }
68}
69
70#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
71pub struct WorkflowSpec {
72 #[serde(default)]
73 pub id: Option<String>,
74 pub goal: String,
75 #[serde(default)]
76 pub description: Option<String>,
77 #[serde(default)]
78 pub budget: BudgetSpec,
79 #[serde(default)]
80 pub permissions: PermissionSpec,
81 #[serde(default)]
82 pub model_policy: ModelPolicy,
83 #[serde(default)]
84 pub promotion_policy: PromotionPolicy,
85 #[serde(default, skip_serializing_if = "Vec::is_empty")]
88 pub gates: Vec<GateSpec>,
89 #[serde(default)]
90 pub nodes: Vec<WorkflowNode>,
91}
92
93impl WorkflowSpec {
94 pub fn validate_for_fleet(&self) -> Result<WorkflowFleetShape, WorkflowFleetLimitError> {
95 self.validate_for_fleet_with_limits(WorkflowFleetLimits::default())
96 }
97
98 pub fn validate_for_fleet_with_limits(
99 &self,
100 limits: WorkflowFleetLimits,
101 ) -> Result<WorkflowFleetShape, WorkflowFleetLimitError> {
102 validate_workflow_nodes(&self.nodes)
103 .map_err(|source| WorkflowFleetLimitError::InvalidWorkflow { source })?;
104 let shape = estimate_fleet_shape(&self.nodes)?;
105 if shape.total_agents > limits.max_total_agents {
106 return Err(WorkflowFleetLimitError::TooManyAgents {
107 total_agents: shape.total_agents,
108 max_total_agents: limits.max_total_agents,
109 });
110 }
111 if shape.max_depth > limits.max_depth {
112 return Err(WorkflowFleetLimitError::RecursionTooDeep {
113 depth: shape.max_depth,
114 max_depth: limits.max_depth,
115 });
116 }
117 Ok(shape)
118 }
119}
120
121#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
122#[serde(tag = "kind", content = "spec", rename_all = "snake_case")]
123pub enum WorkflowNode {
124 BranchSet(BranchSpec),
125 Leaf(LeafSpec),
126 Sequence(SequenceSpec),
127 Reduce(ReduceSpec),
128 TeacherReview(TeacherReviewSpec),
129 LoopUntil(LoopUntilSpec),
130 Cond(CondSpec),
131 Expand(ExpandSpec),
132}
133
134#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
135pub struct BranchSpec {
136 pub id: String,
137 #[serde(default)]
138 pub description: Option<String>,
139 #[serde(default)]
140 pub parallel: bool,
141 #[serde(default)]
142 pub budget: BudgetSpec,
143 #[serde(default)]
144 pub permissions: PermissionSpec,
145 #[serde(default)]
146 pub model_policy: ModelPolicy,
147 #[serde(default)]
148 pub children: Vec<WorkflowNode>,
149}
150
151#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
152pub struct LeafSpec {
153 pub id: String,
154 pub prompt: String,
155 #[serde(default)]
156 pub agent_type: AgentType,
157 #[serde(default, skip_serializing_if = "Option::is_none")]
163 pub role: Option<String>,
164 #[serde(default, skip_serializing_if = "Option::is_none")]
170 pub profile: Option<String>,
171 #[serde(default)]
172 pub mode: TaskMode,
173 #[serde(default)]
174 pub isolation: IsolationMode,
175 #[serde(default)]
176 pub file_scope: Vec<String>,
177 #[serde(default)]
178 pub depends_on_results: Vec<String>,
179 #[serde(default)]
180 pub budget: BudgetSpec,
181 #[serde(default)]
182 pub permissions: PermissionSpec,
183 #[serde(default)]
184 pub model_policy: ModelPolicy,
185}
186
187#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
188pub struct SequenceSpec {
189 pub id: String,
190 #[serde(default)]
191 pub children: Vec<WorkflowNode>,
192}
193
194#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
195pub struct ReduceSpec {
196 pub id: String,
197 #[serde(default)]
198 pub inputs: Vec<String>,
199 pub prompt: String,
200 #[serde(default)]
201 pub model_policy: ModelPolicy,
202}
203
204#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
205pub struct TeacherReviewSpec {
206 pub id: String,
207 #[serde(default)]
208 pub candidates: Vec<String>,
209 #[serde(default)]
210 pub promotion_policy: PromotionPolicy,
211}
212
213#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
214pub struct LoopUntilSpec {
215 pub id: String,
216 pub condition: String,
217 #[serde(default)]
218 pub max_iterations: Option<u32>,
219 #[serde(default)]
220 pub children: Vec<WorkflowNode>,
221}
222
223#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
224pub struct CondSpec {
225 pub id: String,
226 pub condition: String,
227 #[serde(default)]
228 pub then_nodes: Vec<WorkflowNode>,
229 #[serde(default)]
230 pub else_nodes: Vec<WorkflowNode>,
231}
232
233#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
234pub struct ExpandSpec {
235 pub id: String,
236 pub source: String,
237 #[serde(default)]
238 pub max_children: Option<usize>,
239 #[serde(default)]
240 pub template: Option<Box<WorkflowNode>>,
241}
242
243#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
244pub struct BudgetSpec {
245 #[serde(default)]
246 pub max_steps: Option<u32>,
247 #[serde(default)]
248 pub timeout_secs: Option<u64>,
249 #[serde(default)]
250 pub max_parallel: Option<u8>,
251 #[serde(default)]
252 pub max_tokens: Option<u64>,
253}
254
255#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
256pub struct PermissionSpec {
257 #[serde(default)]
258 pub allow_write: bool,
259 #[serde(default)]
260 pub allow_network: bool,
261 #[serde(default)]
264 pub deny_all_tools: bool,
265 #[serde(default)]
266 pub allowed_tools: Vec<String>,
267 #[serde(default)]
268 pub file_scope: Vec<String>,
269}
270
271#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
272pub struct ModelPolicy {
273 #[serde(default)]
274 pub provider: Option<String>,
275 #[serde(default)]
276 pub model: Option<String>,
277 #[serde(default)]
278 pub fallback_models: Vec<String>,
279}
280
281#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
282pub struct PromotionPolicy {
283 #[serde(default)]
284 pub strategy: PromotionStrategy,
285 #[serde(default)]
286 pub require_teacher_review: bool,
287 #[serde(default)]
288 pub min_successful_branches: Option<u32>,
289 #[serde(default)]
290 pub promotion_gate: PromotionGate,
291}
292
293#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
294#[serde(rename_all = "snake_case")]
295pub enum PromotionStrategy {
296 #[default]
297 All,
298 FirstSuccess,
299 BestScore,
300 TeacherSelected,
301}
302
303#[derive(Debug, Clone, PartialEq, Eq)]
304pub struct WorkflowPlan {
305 goal: String,
306 max_concurrent: u8,
307 phases: Vec<PhasePlan>,
308}
309
310impl WorkflowPlan {
311 pub fn from_config(config: &WorkflowConfig) -> Result<Self, WorkflowValidationError> {
312 validate_non_empty("workflow goal", &config.goal)?;
313 if !(1..=20).contains(&config.max_concurrent) {
314 return Err(WorkflowValidationError::InvalidMaxConcurrent {
315 value: config.max_concurrent,
316 });
317 }
318 if config.phases.is_empty() {
319 return Err(WorkflowValidationError::EmptyWorkflow);
320 }
321
322 let mut phase_indices = BTreeMap::new();
323 let mut all_tasks = BTreeMap::new();
324 let mut task_phase = BTreeMap::new();
325
326 for (phase_index, phase) in config.phases.iter().enumerate() {
327 validate_non_empty("phase name", &phase.name)?;
328 if phase.tasks.is_empty() {
329 return Err(WorkflowValidationError::EmptyPhase {
330 phase: phase.name.clone(),
331 });
332 }
333 if phase_indices
334 .insert(phase.name.clone(), phase_index)
335 .is_some()
336 {
337 return Err(WorkflowValidationError::DuplicatePhase {
338 phase: phase.name.clone(),
339 });
340 }
341
342 for task in &phase.tasks {
343 validate_non_empty("task id", &task.id)?;
344 validate_non_empty("task prompt", &task.prompt)?;
345 if all_tasks.insert(task.id.clone(), task).is_some() {
346 return Err(WorkflowValidationError::DuplicateTask {
347 task: task.id.clone(),
348 });
349 }
350 task_phase.insert(task.id.clone(), phase.name.clone());
351 }
352 }
353
354 for phase in &config.phases {
355 for dependency in &phase.depends_on {
356 if dependency == &phase.name || !phase_indices.contains_key(dependency) {
357 return Err(WorkflowValidationError::InvalidPhaseDependency {
358 phase: phase.name.clone(),
359 dependency: dependency.clone(),
360 });
361 }
362 }
363 validate_parallel_write_scope(phase)?;
364 }
365
366 let ordered_phase_names = ordered_phases(config, &phase_indices)?;
367 let phase_order: BTreeMap<_, _> = ordered_phase_names
368 .iter()
369 .enumerate()
370 .map(|(index, phase)| (phase.clone(), index))
371 .collect();
372
373 for phase in &config.phases {
374 for task in &phase.tasks {
375 for dependency in &task.depends_on_results {
376 let Some(dependency_phase) = task_phase.get(dependency) else {
377 return Err(WorkflowValidationError::InvalidTaskResultDependency {
378 task: task.id.clone(),
379 dependency: dependency.clone(),
380 });
381 };
382 if phase_order[dependency_phase] >= phase_order[&phase.name] {
383 return Err(WorkflowValidationError::UnavailableTaskResultDependency {
384 task: task.id.clone(),
385 dependency: dependency.clone(),
386 dependency_phase: dependency_phase.clone(),
387 task_phase: phase.name.clone(),
388 });
389 }
390 }
391 }
392 }
393
394 let phases = ordered_phase_names
395 .iter()
396 .map(|phase_name| {
397 let phase = &config.phases[phase_indices[phase_name]];
398 PhasePlan {
399 name: phase.name.clone(),
400 parallel: phase.parallel,
401 on_failure: phase.on_failure,
402 tasks: phase.tasks.clone(),
403 }
404 })
405 .collect();
406
407 Ok(Self {
408 goal: config.goal.clone(),
409 max_concurrent: config.max_concurrent,
410 phases,
411 })
412 }
413
414 pub fn goal(&self) -> &str {
415 &self.goal
416 }
417
418 pub fn max_concurrent(&self) -> u8 {
419 self.max_concurrent
420 }
421
422 pub fn phases(&self) -> &[PhasePlan] {
423 &self.phases
424 }
425
426 pub fn phase_names(&self) -> impl Iterator<Item = &str> {
427 self.phases.iter().map(|phase| phase.name.as_str())
428 }
429}
430
431pub type WorkflowIr = WorkflowPlan;
432
433#[derive(Debug, Clone, PartialEq, Eq)]
434pub struct PhasePlan {
435 pub name: String,
436 pub parallel: bool,
437 pub on_failure: FailurePolicy,
438 pub tasks: Vec<Task>,
439}
440
441#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
442pub struct Phase {
443 pub name: String,
444 #[serde(default)]
445 pub description: Option<String>,
446 #[serde(default)]
447 pub depends_on: Vec<String>,
448 #[serde(default)]
449 pub parallel: bool,
450 #[serde(default)]
451 pub on_failure: FailurePolicy,
452 #[serde(default)]
453 pub tasks: Vec<Task>,
454}
455
456pub type WorkflowPhase = Phase;
457
458#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
459#[serde(rename_all = "snake_case")]
460pub enum FailurePolicy {
461 #[default]
462 SkipContinue,
463 Abort,
464}
465
466#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
467pub struct Task {
468 pub id: String,
469 pub prompt: String,
470 #[serde(default)]
471 pub agent_type: AgentType,
472 #[serde(default)]
473 pub mode: TaskMode,
474 #[serde(default)]
475 pub isolation: IsolationMode,
476 #[serde(default)]
477 pub file_scope: Vec<String>,
478 #[serde(default)]
479 pub depends_on_results: Vec<String>,
480 #[serde(default)]
481 pub max_steps: Option<u32>,
482 #[serde(default)]
483 pub timeout_secs: Option<u64>,
484}
485
486pub type WorkflowTask = Task;
487pub type WorkflowRole = AgentType;
488
489#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
490#[serde(rename_all = "snake_case")]
491pub enum AgentType {
492 #[default]
493 General,
494 Explore,
495 Plan,
496 Review,
497 Implementer,
498 Verifier,
499}
500
501#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
502#[serde(rename_all = "snake_case")]
503pub enum TaskMode {
504 #[default]
505 ReadOnly,
506 ReadWrite,
507}
508
509#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
510#[serde(rename_all = "snake_case")]
511pub enum IsolationMode {
512 #[default]
518 Auto,
519 Shared,
521 Worktree,
523}
524
525impl IsolationMode {
526 #[must_use]
531 pub fn resolve(self, parallel_write: bool) -> Self {
532 match self {
533 Self::Auto if parallel_write => Self::Worktree,
534 Self::Auto => Self::Shared,
535 other => other,
536 }
537 }
538
539 #[must_use]
541 pub fn wants_worktree(self, parallel_write: bool) -> bool {
542 matches!(self.resolve(parallel_write), Self::Worktree)
543 }
544}
545
546#[must_use]
553pub fn leaf_is_write_capable(spec: &LeafSpec) -> bool {
554 spec.mode == TaskMode::ReadWrite || spec.permissions.allow_write
555}
556
557#[must_use]
560pub fn leaf_wants_worktree(spec: &LeafSpec, parallel: bool) -> bool {
561 let parallel_write = parallel && leaf_is_write_capable(spec);
562 spec.isolation.wants_worktree(parallel_write)
563}
564
565#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
566pub struct BranchResult {
567 pub branch_id: String,
568 pub task_id: String,
569 pub status: WorkflowRunStatus,
570 #[serde(default)]
571 pub usage: WorkflowUsage,
572 #[serde(default)]
573 pub memo_usage: WorkflowMemoUsage,
574 #[serde(default)]
575 pub artifacts: Vec<String>,
576 #[serde(default)]
577 pub notes: Option<String>,
578}
579
580#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
581pub struct LeafResult {
582 pub leaf_id: String,
583 pub task_id: String,
584 #[serde(default, skip_serializing_if = "Option::is_none")]
586 pub role: Option<String>,
587 #[serde(default, skip_serializing_if = "Option::is_none")]
589 pub profile: Option<String>,
590 pub status: WorkflowRunStatus,
591 #[serde(default)]
592 pub usage: WorkflowUsage,
593 #[serde(default)]
594 pub memo_usage: WorkflowMemoUsage,
595 #[serde(default)]
596 pub output: Option<String>,
597 #[serde(default)]
598 pub artifacts: Vec<String>,
599 #[serde(default, skip_serializing_if = "Option::is_none")]
601 pub schema_error: Option<String>,
602}
603
604#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
605pub struct WorkflowUsage {
606 #[serde(default)]
607 pub input_tokens: u64,
608 #[serde(default)]
609 pub output_tokens: u64,
610 #[serde(default)]
611 pub cost_microusd: u64,
612}
613
614impl WorkflowUsage {
615 #[must_use]
616 pub fn total_tokens(self) -> u64 {
617 self.input_tokens.saturating_add(self.output_tokens)
618 }
619
620 pub(crate) fn add_assign(&mut self, other: Self) {
621 self.input_tokens = self.input_tokens.saturating_add(other.input_tokens);
622 self.output_tokens = self.output_tokens.saturating_add(other.output_tokens);
623 self.cost_microusd = self.cost_microusd.saturating_add(other.cost_microusd);
624 }
625}
626
627#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
628pub struct WorkflowMemoUsage {
629 #[serde(default)]
630 pub armh_hits: u64,
631 #[serde(default)]
632 pub armh_misses: u64,
633 #[serde(default)]
634 pub armh_saved_estimated_tokens: u64,
635 #[serde(default)]
636 pub provider_prompt_cache_hits: u64,
637 #[serde(default)]
638 pub provider_prompt_cache_misses: u64,
639}
640
641impl WorkflowMemoUsage {
642 pub(crate) fn add_assign(&mut self, other: Self) {
643 self.armh_hits = self.armh_hits.saturating_add(other.armh_hits);
644 self.armh_misses = self.armh_misses.saturating_add(other.armh_misses);
645 self.armh_saved_estimated_tokens = self
646 .armh_saved_estimated_tokens
647 .saturating_add(other.armh_saved_estimated_tokens);
648 self.provider_prompt_cache_hits = self
649 .provider_prompt_cache_hits
650 .saturating_add(other.provider_prompt_cache_hits);
651 self.provider_prompt_cache_misses = self
652 .provider_prompt_cache_misses
653 .saturating_add(other.provider_prompt_cache_misses);
654 }
655}
656
657#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
658pub struct ControlNodeResult {
659 pub node_id: String,
660 pub kind: ControlNodeKind,
661 pub status: WorkflowRunStatus,
662 #[serde(default)]
663 pub selected_children: Vec<String>,
664 #[serde(default)]
665 pub summary: Option<String>,
666}
667
668#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
669#[serde(rename_all = "snake_case")]
670pub enum WorkflowRunStatus {
671 #[default]
672 Pending,
673 Running,
674 Succeeded,
675 Failed,
676 Cancelled,
677 BudgetExceeded,
678 ReplayDiverged,
679}
680
681#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
682#[serde(rename_all = "snake_case")]
683pub enum ControlNodeKind {
684 BranchSet,
685 Leaf,
686 Sequence,
687 Reduce,
688 TeacherReview,
689 LoopUntil,
690 Cond,
691 Expand,
692}
693
694#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
695pub struct WorkflowExecution {
696 pub status: WorkflowRunStatus,
697 #[serde(default)]
698 pub usage: WorkflowUsage,
699 #[serde(default)]
700 pub memo_usage: WorkflowMemoUsage,
701 #[serde(default)]
702 pub leaf_results: Vec<LeafResult>,
703 #[serde(default)]
704 pub branch_results: Vec<BranchResult>,
705 #[serde(default)]
706 pub control_node_results: Vec<ControlNodeResult>,
707}
708
709impl Default for WorkflowExecution {
710 fn default() -> Self {
711 Self {
712 status: WorkflowRunStatus::Succeeded,
713 usage: WorkflowUsage::default(),
714 memo_usage: WorkflowMemoUsage::default(),
715 leaf_results: Vec::new(),
716 branch_results: Vec::new(),
717 control_node_results: Vec::new(),
718 }
719 }
720}
721
722impl WorkflowExecution {
723 pub fn mark_failed(&mut self) {
724 self.status = WorkflowRunStatus::Failed;
725 }
726
727 pub fn mark_cancelled(&mut self) {
728 self.status = WorkflowRunStatus::Cancelled;
729 }
730
731 pub fn mark_budget_exceeded(&mut self) {
732 self.status = WorkflowRunStatus::BudgetExceeded;
733 }
734
735 pub(crate) fn mark_replay_diverged(&mut self) {
736 self.status = WorkflowRunStatus::ReplayDiverged;
737 }
738
739 fn should_stop_mock_execution(&self) -> bool {
740 matches!(
741 self.status,
742 WorkflowRunStatus::Cancelled | WorkflowRunStatus::BudgetExceeded
743 )
744 }
745}
746
747#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
748pub struct MockLeafOutcome {
749 pub status: WorkflowRunStatus,
750 #[serde(default)]
751 pub usage: WorkflowUsage,
752 #[serde(default)]
753 pub memo_usage: WorkflowMemoUsage,
754 #[serde(default)]
755 pub output: Option<String>,
756 #[serde(default)]
757 pub artifacts: Vec<String>,
758}
759
760impl MockLeafOutcome {
761 pub fn succeeded(output: impl Into<String>) -> Self {
762 Self {
763 status: WorkflowRunStatus::Succeeded,
764 usage: WorkflowUsage::default(),
765 memo_usage: WorkflowMemoUsage::default(),
766 output: Some(output.into()),
767 artifacts: Vec::new(),
768 }
769 }
770
771 pub fn failed(output: impl Into<String>) -> Self {
772 Self {
773 status: WorkflowRunStatus::Failed,
774 usage: WorkflowUsage::default(),
775 memo_usage: WorkflowMemoUsage::default(),
776 output: Some(output.into()),
777 artifacts: Vec::new(),
778 }
779 }
780
781 pub fn with_usage(mut self, usage: WorkflowUsage) -> Self {
782 self.usage = usage;
783 self
784 }
785
786 pub fn with_memo_usage(mut self, memo_usage: WorkflowMemoUsage) -> Self {
787 self.memo_usage = memo_usage;
788 self
789 }
790}
791
792#[derive(Debug, Default, Clone)]
793pub struct MockWorkflowExecutor {
794 leaf_outcomes: BTreeMap<String, MockLeafOutcome>,
795 predicate_results: BTreeMap<String, Vec<bool>>,
796 generated_nodes: BTreeMap<String, Vec<WorkflowNode>>,
797 cancelled: bool,
798 max_leaf_steps: Option<u32>,
799 leaf_steps_executed: u32,
800 max_leaf_tokens: Option<u64>,
801 leaf_tokens_used: u64,
802}
803
804impl MockWorkflowExecutor {
805 pub fn new() -> Self {
806 Self::default()
807 }
808
809 pub fn with_leaf_outcome(
810 mut self,
811 leaf_id: impl Into<String>,
812 outcome: MockLeafOutcome,
813 ) -> Self {
814 self.leaf_outcomes.insert(leaf_id.into(), outcome);
815 self
816 }
817
818 pub fn with_predicate_results(
819 mut self,
820 node_id: impl Into<String>,
821 results: Vec<bool>,
822 ) -> Self {
823 self.predicate_results.insert(node_id.into(), results);
824 self
825 }
826
827 pub fn with_generated_nodes(
828 mut self,
829 node_id: impl Into<String>,
830 nodes: Vec<WorkflowNode>,
831 ) -> Self {
832 self.generated_nodes.insert(node_id.into(), nodes);
833 self
834 }
835
836 pub fn with_cancelled(mut self) -> Self {
837 self.cancelled = true;
838 self
839 }
840
841 pub fn with_max_leaf_steps(mut self, max_leaf_steps: u32) -> Self {
842 self.max_leaf_steps = Some(max_leaf_steps);
843 self
844 }
845
846 pub fn with_max_leaf_tokens(mut self, max_leaf_tokens: u64) -> Self {
847 self.max_leaf_tokens = Some(max_leaf_tokens);
848 self
849 }
850
851 pub fn run(
852 &mut self,
853 spec: &WorkflowSpec,
854 ) -> Result<WorkflowExecution, WorkflowExecutionError> {
855 validate_workflow_nodes(&spec.nodes)?;
856 let mut execution = WorkflowExecution::default();
857 self.execute_nodes(&spec.nodes, &mut execution)?;
858 Ok(execution)
859 }
860
861 fn execute_nodes(
862 &mut self,
863 nodes: &[WorkflowNode],
864 execution: &mut WorkflowExecution,
865 ) -> Result<(), WorkflowExecutionError> {
866 for node in nodes {
867 if execution.should_stop_mock_execution() {
868 break;
869 }
870 self.execute_node(node, execution)?;
871 }
872 Ok(())
873 }
874
875 fn execute_node(
876 &mut self,
877 node: &WorkflowNode,
878 execution: &mut WorkflowExecution,
879 ) -> Result<(), WorkflowExecutionError> {
880 match node {
881 WorkflowNode::BranchSet(spec) => self.execute_branch_set(spec, execution),
882 WorkflowNode::Leaf(spec) => {
883 self.execute_leaf(spec, execution);
884 Ok(())
885 }
886 WorkflowNode::Sequence(spec) => {
887 self.execute_nodes(&spec.children, execution)?;
888 execution.control_node_results.push(ControlNodeResult {
889 node_id: spec.id.clone(),
890 kind: ControlNodeKind::Sequence,
891 status: execution.status,
892 selected_children: spec.children.iter().map(node_id).collect(),
893 summary: Some("sequence executed in declaration order".to_string()),
894 });
895 Ok(())
896 }
897 WorkflowNode::Reduce(spec) => {
898 execution.control_node_results.push(ControlNodeResult {
899 node_id: spec.id.clone(),
900 kind: ControlNodeKind::Reduce,
901 status: WorkflowRunStatus::Succeeded,
902 selected_children: spec.inputs.clone(),
903 summary: Some(spec.prompt.clone()),
904 });
905 Ok(())
906 }
907 WorkflowNode::TeacherReview(spec) => {
908 execution.control_node_results.push(ControlNodeResult {
909 node_id: spec.id.clone(),
910 kind: ControlNodeKind::TeacherReview,
911 status: WorkflowRunStatus::Succeeded,
912 selected_children: spec.candidates.clone(),
913 summary: Some(
914 "teacher review scaffold selected declared candidates".to_string(),
915 ),
916 });
917 Ok(())
918 }
919 WorkflowNode::LoopUntil(spec) => self.execute_loop_until(spec, execution),
920 WorkflowNode::Cond(spec) => self.execute_cond(spec, execution),
921 WorkflowNode::Expand(spec) => self.execute_expand(spec, execution),
922 }
923 }
924
925 fn execute_branch_set(
926 &mut self,
927 spec: &BranchSpec,
928 execution: &mut WorkflowExecution,
929 ) -> Result<(), WorkflowExecutionError> {
930 let before = execution.leaf_results.len();
931 self.execute_nodes(&spec.children, execution)?;
932 let status = aggregate_mock_status(&execution.leaf_results[before..]);
933 let mut usage = WorkflowUsage::default();
934 let mut memo_usage = WorkflowMemoUsage::default();
935 for result in &execution.leaf_results[before..] {
936 usage.add_assign(result.usage);
937 memo_usage.add_assign(result.memo_usage);
938 }
939 mark_execution_for_status(execution, status);
940 execution.branch_results.push(BranchResult {
941 branch_id: spec.id.clone(),
942 task_id: spec.id.clone(),
943 status,
944 usage,
945 memo_usage,
946 artifacts: Vec::new(),
947 notes: Some("mock branch set executed without runtime fanout".to_string()),
948 });
949 execution.control_node_results.push(ControlNodeResult {
950 node_id: spec.id.clone(),
951 kind: ControlNodeKind::BranchSet,
952 status,
953 selected_children: spec.children.iter().map(node_id).collect(),
954 summary: Some("branch set scaffold executed children deterministically".to_string()),
955 });
956 Ok(())
957 }
958
959 fn execute_leaf(&mut self, spec: &LeafSpec, execution: &mut WorkflowExecution) {
960 let outcome = self.mock_leaf_outcome(spec);
961 mark_execution_for_status(execution, outcome.status);
962 execution.usage.add_assign(outcome.usage);
963 execution.memo_usage.add_assign(outcome.memo_usage);
964 execution.leaf_results.push(LeafResult {
965 leaf_id: spec.id.clone(),
966 task_id: spec.id.clone(),
967 role: spec.role.clone(),
968 profile: spec.profile.clone(),
969 status: outcome.status,
970 usage: outcome.usage,
971 memo_usage: outcome.memo_usage,
972 output: outcome.output,
973 artifacts: outcome.artifacts,
974 schema_error: None,
975 });
976 }
977
978 fn execute_loop_until(
979 &mut self,
980 spec: &LoopUntilSpec,
981 execution: &mut WorkflowExecution,
982 ) -> Result<(), WorkflowExecutionError> {
983 let max_iterations = spec.max_iterations.unwrap_or(1).max(1);
984 let mut iterations = 0;
985 let mut passed = false;
986 while iterations < max_iterations {
987 if execution.should_stop_mock_execution() {
988 break;
989 }
990 iterations += 1;
991 self.execute_nodes(&spec.children, execution)?;
992 if execution.should_stop_mock_execution() {
993 break;
994 }
995 if self.next_predicate_result(&spec.id) {
996 passed = true;
997 break;
998 }
999 }
1000 let status = if execution.should_stop_mock_execution() {
1001 execution.status
1002 } else if passed {
1003 WorkflowRunStatus::Succeeded
1004 } else {
1005 WorkflowRunStatus::Failed
1006 };
1007 mark_execution_for_status(execution, status);
1008 execution.control_node_results.push(ControlNodeResult {
1009 node_id: spec.id.clone(),
1010 kind: ControlNodeKind::LoopUntil,
1011 status,
1012 selected_children: spec.children.iter().map(node_id).collect(),
1013 summary: Some(format!("loop_until iterations={iterations}")),
1014 });
1015 Ok(())
1016 }
1017
1018 fn execute_cond(
1019 &mut self,
1020 spec: &CondSpec,
1021 execution: &mut WorkflowExecution,
1022 ) -> Result<(), WorkflowExecutionError> {
1023 let passed = self.next_predicate_result(&spec.id);
1024 let selected_nodes = if passed {
1025 &spec.then_nodes
1026 } else {
1027 &spec.else_nodes
1028 };
1029 self.execute_nodes(selected_nodes, execution)?;
1030 let status = if execution.should_stop_mock_execution() {
1031 execution.status
1032 } else {
1033 WorkflowRunStatus::Succeeded
1034 };
1035 execution.control_node_results.push(ControlNodeResult {
1036 node_id: spec.id.clone(),
1037 kind: ControlNodeKind::Cond,
1038 status,
1039 selected_children: selected_nodes.iter().map(node_id).collect(),
1040 summary: Some(format!("predicate_result={passed}")),
1041 });
1042 Ok(())
1043 }
1044
1045 fn execute_expand(
1046 &mut self,
1047 spec: &ExpandSpec,
1048 execution: &mut WorkflowExecution,
1049 ) -> Result<(), WorkflowExecutionError> {
1050 let mut nodes = self.generated_nodes.remove(&spec.id).unwrap_or_default();
1051 if let Some(max_children) = spec.max_children {
1052 nodes.truncate(max_children);
1053 }
1054 validate_workflow_node_shapes(&nodes)?;
1055 self.execute_nodes(&nodes, execution)?;
1056 let status = if execution.should_stop_mock_execution() {
1057 execution.status
1058 } else {
1059 WorkflowRunStatus::Succeeded
1060 };
1061 execution.control_node_results.push(ControlNodeResult {
1062 node_id: spec.id.clone(),
1063 kind: ControlNodeKind::Expand,
1064 status,
1065 selected_children: nodes.iter().map(node_id).collect(),
1066 summary: Some(format!("expanded_from={}", spec.source)),
1067 });
1068 Ok(())
1069 }
1070
1071 fn mock_leaf_outcome(&mut self, spec: &LeafSpec) -> MockLeafOutcome {
1072 if self.cancelled {
1073 return MockLeafOutcome {
1074 status: WorkflowRunStatus::Cancelled,
1075 usage: WorkflowUsage::default(),
1076 memo_usage: WorkflowMemoUsage::default(),
1077 output: Some("mock workflow cancelled before leaf execution".to_string()),
1078 artifacts: Vec::new(),
1079 };
1080 }
1081 if self.max_leaf_steps == Some(self.leaf_steps_executed) || spec.budget.max_steps == Some(0)
1082 {
1083 return MockLeafOutcome {
1084 status: WorkflowRunStatus::BudgetExceeded,
1085 usage: WorkflowUsage::default(),
1086 memo_usage: WorkflowMemoUsage::default(),
1087 output: Some("mock workflow leaf step budget exhausted".to_string()),
1088 artifacts: Vec::new(),
1089 };
1090 }
1091 if self
1092 .max_leaf_tokens
1093 .is_some_and(|max| self.leaf_tokens_used >= max)
1094 || spec.budget.max_tokens == Some(0)
1095 {
1096 return MockLeafOutcome {
1097 status: WorkflowRunStatus::BudgetExceeded,
1098 usage: WorkflowUsage::default(),
1099 memo_usage: WorkflowMemoUsage::default(),
1100 output: Some("mock workflow leaf token budget exhausted".to_string()),
1101 artifacts: Vec::new(),
1102 };
1103 }
1104 self.leaf_steps_executed = self.leaf_steps_executed.saturating_add(1);
1105 let outcome = self
1106 .leaf_outcomes
1107 .remove(&spec.id)
1108 .unwrap_or_else(|| MockLeafOutcome::succeeded(format!("mock leaf {}", spec.id)));
1109 let tokens = outcome.usage.total_tokens();
1110 if let Some(per_leaf_token_cap) = spec.budget.max_tokens
1111 && tokens > per_leaf_token_cap
1112 {
1113 return MockLeafOutcome {
1114 status: WorkflowRunStatus::BudgetExceeded,
1115 usage: outcome.usage,
1116 memo_usage: outcome.memo_usage,
1117 output: Some(format!(
1118 "mock workflow leaf token budget exhausted ({tokens} > {per_leaf_token_cap})"
1119 )),
1120 artifacts: outcome.artifacts,
1121 };
1122 }
1123 self.leaf_tokens_used = self.leaf_tokens_used.saturating_add(tokens);
1124 outcome
1125 }
1126
1127 fn next_predicate_result(&mut self, node_id: &str) -> bool {
1128 let Some(results) = self.predicate_results.get_mut(node_id) else {
1129 return false;
1130 };
1131 if results.is_empty() {
1132 return false;
1133 }
1134 results.remove(0)
1135 }
1136}
1137
1138fn aggregate_mock_status(results: &[LeafResult]) -> WorkflowRunStatus {
1139 if results
1140 .iter()
1141 .any(|result| result.status == WorkflowRunStatus::Cancelled)
1142 {
1143 WorkflowRunStatus::Cancelled
1144 } else if results
1145 .iter()
1146 .any(|result| result.status == WorkflowRunStatus::BudgetExceeded)
1147 {
1148 WorkflowRunStatus::BudgetExceeded
1149 } else if results
1150 .iter()
1151 .any(|result| result.status != WorkflowRunStatus::Succeeded)
1152 {
1153 WorkflowRunStatus::Failed
1154 } else {
1155 WorkflowRunStatus::Succeeded
1156 }
1157}
1158
1159fn mark_execution_for_status(execution: &mut WorkflowExecution, status: WorkflowRunStatus) {
1160 match status {
1161 WorkflowRunStatus::Succeeded | WorkflowRunStatus::Pending | WorkflowRunStatus::Running => {}
1162 WorkflowRunStatus::Failed => execution.mark_failed(),
1163 WorkflowRunStatus::Cancelled => execution.mark_cancelled(),
1164 WorkflowRunStatus::BudgetExceeded => execution.mark_budget_exceeded(),
1165 WorkflowRunStatus::ReplayDiverged => execution.mark_replay_diverged(),
1166 }
1167}
1168
1169#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
1170pub struct BranchCandidate {
1171 pub branch_id: String,
1172 pub status: WorkflowRunStatus,
1173 pub score: u32,
1174 pub cost: u64,
1175 #[serde(default)]
1176 pub diversity_key: Option<String>,
1177}
1178
1179#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
1180#[serde(rename_all = "snake_case")]
1181pub enum TeacherCandidateKind {
1182 Note,
1183 WorkflowRecipe,
1184 SkillPatch,
1185 RegressionTest,
1186 CachePolicyPatch,
1187 BranchHeuristic,
1188 AuthoringPromptPatch,
1189}
1190
1191#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
1192#[serde(rename_all = "snake_case")]
1193pub enum TeacherCandidateStatus {
1194 #[default]
1195 Proposed,
1196 Accepted,
1197 Rejected,
1198 Promoted,
1199}
1200
1201#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
1202pub struct TeacherCandidate {
1203 pub candidate_id: String,
1204 pub kind: TeacherCandidateKind,
1205 #[serde(default)]
1206 pub status: TeacherCandidateStatus,
1207 pub source_node_id: String,
1208 #[serde(default)]
1209 pub source_branch_id: Option<String>,
1210 pub summary: String,
1211 #[serde(default)]
1212 pub evidence: Vec<String>,
1213 #[serde(default)]
1214 pub replay_results: Vec<StudentReplayResult>,
1215}
1216
1217#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
1218pub struct StudentReplayMetrics {
1219 #[serde(default)]
1220 pub score: i32,
1221 #[serde(default)]
1222 pub cost_microusd: u64,
1223}
1224
1225#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
1226pub struct StudentReplayTestResult {
1227 pub name: String,
1228 pub passed: bool,
1229}
1230
1231#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
1232pub struct StudentReplayResult {
1233 pub trace_id: String,
1234 pub candidate_id: String,
1235 pub baseline: StudentReplayMetrics,
1236 pub candidate: StudentReplayMetrics,
1237 #[serde(default)]
1238 pub required_tests: Vec<StudentReplayTestResult>,
1239 #[serde(default)]
1240 pub policy_violations: Vec<String>,
1241 #[serde(default)]
1242 pub stale: bool,
1243 #[serde(default)]
1244 pub notes: Option<String>,
1245}
1246
1247#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
1248pub struct PromotionGate {
1249 #[serde(default = "default_min_replay_score_delta")]
1250 pub min_score_delta: i32,
1251 #[serde(default)]
1252 pub max_cost_delta_microusd: Option<i64>,
1253 #[serde(default = "default_true")]
1254 pub require_all_tests_pass: bool,
1255 #[serde(default = "default_true")]
1256 pub reject_policy_violations: bool,
1257 #[serde(default = "default_true")]
1258 pub reject_stale_replay: bool,
1259}
1260
1261impl Default for PromotionGate {
1262 fn default() -> Self {
1263 Self {
1264 min_score_delta: default_min_replay_score_delta(),
1265 max_cost_delta_microusd: None,
1266 require_all_tests_pass: true,
1267 reject_policy_violations: true,
1268 reject_stale_replay: true,
1269 }
1270 }
1271}
1272
1273impl PromotionGate {
1274 pub fn evaluate_candidate(&self, candidate: &TeacherCandidate) -> PromotionGateDecision {
1275 let Some(replay) = candidate.replay_results.last() else {
1276 return PromotionGateDecision {
1277 candidate_id: candidate.candidate_id.clone(),
1278 status: TeacherCandidateStatus::Rejected,
1279 score_delta: 0,
1280 cost_delta_microusd: 0,
1281 reasons: vec!["no student replay result recorded".to_string()],
1282 };
1283 };
1284 self.evaluate_replay(&candidate.candidate_id, replay)
1285 }
1286
1287 pub fn evaluate_replay(
1288 &self,
1289 candidate_id: &str,
1290 replay: &StudentReplayResult,
1291 ) -> PromotionGateDecision {
1292 let score_delta = replay.score_delta();
1293 let cost_delta_microusd = replay.cost_delta_microusd();
1294 let mut reasons = Vec::new();
1295
1296 if score_delta < self.min_score_delta {
1297 reasons.push(format!(
1298 "score delta {score_delta} is below required {}",
1299 self.min_score_delta
1300 ));
1301 }
1302 if let Some(max_cost_delta) = self.max_cost_delta_microusd
1303 && cost_delta_microusd > max_cost_delta
1304 {
1305 reasons.push(format!(
1306 "cost delta {cost_delta_microusd} exceeds allowed {max_cost_delta}"
1307 ));
1308 }
1309 if self.require_all_tests_pass {
1310 for test in replay.required_tests.iter().filter(|test| !test.passed) {
1311 reasons.push(format!("required test `{}` failed", test.name));
1312 }
1313 }
1314 if self.reject_policy_violations {
1315 for violation in &replay.policy_violations {
1316 reasons.push(format!("policy violation: {violation}"));
1317 }
1318 }
1319 if self.reject_stale_replay && replay.stale {
1320 reasons.push("student replay result is stale".to_string());
1321 }
1322
1323 let status = if reasons.is_empty() {
1324 TeacherCandidateStatus::Promoted
1325 } else {
1326 TeacherCandidateStatus::Rejected
1327 };
1328 PromotionGateDecision {
1329 candidate_id: candidate_id.to_string(),
1330 status,
1331 score_delta,
1332 cost_delta_microusd,
1333 reasons,
1334 }
1335 }
1336}
1337
1338#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
1339pub struct PromotionGateDecision {
1340 pub candidate_id: String,
1341 pub status: TeacherCandidateStatus,
1342 pub score_delta: i32,
1343 pub cost_delta_microusd: i64,
1344 #[serde(default)]
1345 pub reasons: Vec<String>,
1346}
1347
1348impl PromotionGateDecision {
1349 pub fn promoted(&self) -> bool {
1350 self.status == TeacherCandidateStatus::Promoted
1351 }
1352}
1353
1354impl StudentReplayResult {
1355 pub fn score_delta(&self) -> i32 {
1356 self.candidate.score.saturating_sub(self.baseline.score)
1357 }
1358
1359 pub fn cost_delta_microusd(&self) -> i64 {
1360 signed_u64_delta(self.candidate.cost_microusd, self.baseline.cost_microusd)
1361 }
1362}
1363
1364#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
1365pub struct TeacherReviewReport {
1366 pub review_node_id: String,
1367 #[serde(default)]
1368 pub candidates: Vec<TeacherCandidate>,
1369}
1370
1371impl TeacherReviewReport {
1372 pub fn from_execution(review: &TeacherReviewSpec, execution: &WorkflowExecution) -> Self {
1373 let candidates = teacher_candidates_from_execution(review, execution);
1374 Self {
1375 review_node_id: review.id.clone(),
1376 candidates,
1377 }
1378 }
1379}
1380
1381pub fn teacher_candidates_from_execution(
1382 review: &TeacherReviewSpec,
1383 execution: &WorkflowExecution,
1384) -> Vec<TeacherCandidate> {
1385 let mut candidates = Vec::new();
1386 for source in &review.candidates {
1387 if let Some(branch) = execution
1388 .branch_results
1389 .iter()
1390 .find(|branch| branch.branch_id == *source || branch.task_id == *source)
1391 {
1392 candidates.push(teacher_candidate_from_branch(review, branch));
1393 continue;
1394 }
1395 if let Some(leaf) = execution
1396 .leaf_results
1397 .iter()
1398 .find(|leaf| leaf.leaf_id == *source || leaf.task_id == *source)
1399 {
1400 candidates.push(teacher_candidate_from_leaf(review, leaf));
1401 continue;
1402 }
1403 if let Some(control) = execution
1404 .control_node_results
1405 .iter()
1406 .find(|control| control.node_id == *source)
1407 {
1408 candidates.push(teacher_candidate_from_control(review, control));
1409 }
1410 }
1411 candidates
1412}
1413
1414fn teacher_candidate_from_branch(
1415 review: &TeacherReviewSpec,
1416 branch: &BranchResult,
1417) -> TeacherCandidate {
1418 let kind =
1419 if branch.memo_usage.armh_hits > 0 || branch.memo_usage.provider_prompt_cache_hits > 0 {
1420 TeacherCandidateKind::CachePolicyPatch
1421 } else if branch.status == WorkflowRunStatus::Succeeded {
1422 TeacherCandidateKind::WorkflowRecipe
1423 } else {
1424 TeacherCandidateKind::BranchHeuristic
1425 };
1426 let mut evidence = vec![format!("status={:?}", branch.status)];
1427 if branch.usage.total_tokens() > 0 || branch.usage.cost_microusd > 0 {
1428 evidence.push(format!(
1429 "tokens={}, cost_microusd={}",
1430 branch.usage.total_tokens(),
1431 branch.usage.cost_microusd
1432 ));
1433 }
1434 if branch.memo_usage.armh_hits > 0 || branch.memo_usage.provider_prompt_cache_hits > 0 {
1435 evidence.push(format!(
1436 "armh_hits={}, provider_prompt_cache_hits={}",
1437 branch.memo_usage.armh_hits, branch.memo_usage.provider_prompt_cache_hits
1438 ));
1439 }
1440 if let Some(notes) = branch.notes.as_deref() {
1441 evidence.push(format!("notes={notes}"));
1442 }
1443 TeacherCandidate {
1444 candidate_id: format!("{}:{}", review.id, branch.branch_id),
1445 kind,
1446 status: TeacherCandidateStatus::Proposed,
1447 source_node_id: branch.task_id.clone(),
1448 source_branch_id: Some(branch.branch_id.clone()),
1449 summary: format!(
1450 "TeacherReview candidate from branch `{}` with {:?} status.",
1451 branch.branch_id, branch.status
1452 ),
1453 evidence,
1454 replay_results: Vec::new(),
1455 }
1456}
1457
1458fn teacher_candidate_from_leaf(review: &TeacherReviewSpec, leaf: &LeafResult) -> TeacherCandidate {
1459 let kind = if leaf.status == WorkflowRunStatus::Failed {
1460 TeacherCandidateKind::RegressionTest
1461 } else if leaf.memo_usage.armh_hits > 0 || leaf.memo_usage.provider_prompt_cache_hits > 0 {
1462 TeacherCandidateKind::CachePolicyPatch
1463 } else {
1464 TeacherCandidateKind::Note
1465 };
1466 let mut evidence = vec![format!("status={:?}", leaf.status)];
1467 if let Some(output) = leaf.output.as_deref() {
1468 evidence.push(format!("output={}", truncate_evidence(output)));
1469 }
1470 TeacherCandidate {
1471 candidate_id: format!("{}:{}", review.id, leaf.leaf_id),
1472 kind,
1473 status: TeacherCandidateStatus::Proposed,
1474 source_node_id: leaf.leaf_id.clone(),
1475 source_branch_id: None,
1476 summary: format!(
1477 "TeacherReview candidate from leaf `{}` with {:?} status.",
1478 leaf.leaf_id, leaf.status
1479 ),
1480 evidence,
1481 replay_results: Vec::new(),
1482 }
1483}
1484
1485fn teacher_candidate_from_control(
1486 review: &TeacherReviewSpec,
1487 control: &ControlNodeResult,
1488) -> TeacherCandidate {
1489 let mut evidence = vec![format!("status={:?}", control.status)];
1490 if !control.selected_children.is_empty() {
1491 evidence.push(format!(
1492 "selected_children={}",
1493 control.selected_children.join(",")
1494 ));
1495 }
1496 if let Some(summary) = control.summary.as_deref() {
1497 evidence.push(format!("summary={}", truncate_evidence(summary)));
1498 }
1499 TeacherCandidate {
1500 candidate_id: format!("{}:{}", review.id, control.node_id),
1501 kind: TeacherCandidateKind::AuthoringPromptPatch,
1502 status: TeacherCandidateStatus::Proposed,
1503 source_node_id: control.node_id.clone(),
1504 source_branch_id: None,
1505 summary: format!(
1506 "TeacherReview candidate from control node `{}` ({:?}).",
1507 control.node_id, control.kind
1508 ),
1509 evidence,
1510 replay_results: Vec::new(),
1511 }
1512}
1513
1514fn default_min_replay_score_delta() -> i32 {
1515 1
1516}
1517
1518fn default_true() -> bool {
1519 true
1520}
1521
1522fn signed_u64_delta(candidate: u64, baseline: u64) -> i64 {
1523 if candidate >= baseline {
1524 i64::try_from(candidate - baseline).unwrap_or(i64::MAX)
1525 } else {
1526 -i64::try_from(baseline - candidate).unwrap_or(i64::MAX)
1527 }
1528}
1529
1530fn truncate_evidence(value: &str) -> String {
1531 const MAX_EVIDENCE_CHARS: usize = 240;
1532 if value.chars().count() <= MAX_EVIDENCE_CHARS {
1533 return value.to_string();
1534 }
1535 let mut truncated = value
1536 .chars()
1537 .take(MAX_EVIDENCE_CHARS.saturating_sub(1))
1538 .collect::<String>();
1539 truncated.push_str("...");
1540 truncated
1541}
1542
1543#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
1544pub struct BranchTournament {
1545 #[serde(default)]
1546 pub min_score: u32,
1547}
1548
1549impl BranchTournament {
1550 pub fn select(&self, candidates: &[BranchCandidate]) -> Option<BranchCandidate> {
1551 candidates
1552 .iter()
1553 .filter(|candidate| {
1554 candidate.status == WorkflowRunStatus::Succeeded
1555 && candidate.score >= self.min_score
1556 })
1557 .min_by_key(|candidate| (candidate.cost, std::cmp::Reverse(candidate.score)))
1558 .cloned()
1559 }
1560}
1561
1562#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
1563pub struct ParetoFrontier {
1564 #[serde(default = "default_frontier_limit")]
1565 pub max_items: usize,
1566}
1567
1568impl Default for ParetoFrontier {
1569 fn default() -> Self {
1570 Self {
1571 max_items: default_frontier_limit(),
1572 }
1573 }
1574}
1575
1576impl ParetoFrontier {
1577 pub fn select(&self, candidates: &[BranchCandidate]) -> Vec<BranchCandidate> {
1578 let mut frontier: Vec<_> = candidates
1579 .iter()
1580 .filter(|candidate| candidate.status == WorkflowRunStatus::Succeeded)
1581 .filter(|candidate| {
1582 !candidates.iter().any(|other| {
1583 other.status == WorkflowRunStatus::Succeeded
1584 && other.score >= candidate.score
1585 && other.cost <= candidate.cost
1586 && (other.score > candidate.score || other.cost < candidate.cost)
1587 })
1588 })
1589 .cloned()
1590 .collect();
1591 frontier.sort_by_key(|candidate| (std::cmp::Reverse(candidate.score), candidate.cost));
1592 frontier.truncate(self.max_items.max(1));
1593 frontier
1594 }
1595}
1596
1597#[derive(Debug, Clone, PartialEq, Eq, Error)]
1598pub enum WorkflowExecutionError {
1599 #[error("{kind} node id must not be empty")]
1600 EmptyNodeId { kind: &'static str },
1601 #[error("leaf `{leaf}` prompt must not be empty")]
1602 EmptyLeafPrompt { leaf: String },
1603 #[error(
1604 "leaf `{leaf}` profile `{profile}` must be a non-empty token without whitespace, quotes, or `=`"
1605 )]
1606 InvalidLeafProfile { leaf: String, profile: String },
1607 #[error(
1608 "leaf `{leaf}` role `{role}` must be a non-empty token without whitespace, quotes, or `=`"
1609 )]
1610 InvalidLeafRole { leaf: String, role: String },
1611 #[error("duplicate workflow node `{node}`")]
1612 DuplicateNodeId { node: String },
1613 #[error("workflow node `{node}` has unknown {field} reference `{reference}`")]
1614 UnknownNodeReference {
1615 node: String,
1616 field: &'static str,
1617 reference: String,
1618 },
1619}
1620
1621#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1622pub struct WorkflowFleetLimits {
1623 pub max_total_agents: usize,
1624 pub max_depth: usize,
1625}
1626
1627impl Default for WorkflowFleetLimits {
1628 fn default() -> Self {
1629 Self {
1630 max_total_agents: DEFAULT_FLEET_WORKFLOW_MAX_AGENTS,
1631 max_depth: DEFAULT_FLEET_WORKFLOW_MAX_DEPTH,
1632 }
1633 }
1634}
1635
1636#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1637pub struct WorkflowFleetShape {
1638 pub total_agents: usize,
1639 pub max_depth: usize,
1640}
1641
1642impl WorkflowFleetShape {
1643 fn add(self, other: Self) -> Self {
1644 Self {
1645 total_agents: self.total_agents.saturating_add(other.total_agents),
1646 max_depth: self.max_depth.max(other.max_depth),
1647 }
1648 }
1649
1650 fn repeat(self, times: usize) -> Self {
1651 Self {
1652 total_agents: self.total_agents.saturating_mul(times),
1653 max_depth: self.max_depth,
1654 }
1655 }
1656}
1657
1658#[derive(Debug, Clone, PartialEq, Eq, Error)]
1659pub enum WorkflowFleetLimitError {
1660 #[error("workflow IR is invalid for Fleet: {source}")]
1661 InvalidWorkflow {
1662 #[from]
1663 source: WorkflowExecutionError,
1664 },
1665 #[error(
1666 "workflow would launch {total_agents} agents; Fleet Workflow limit is {max_total_agents}"
1667 )]
1668 TooManyAgents {
1669 total_agents: usize,
1670 max_total_agents: usize,
1671 },
1672 #[error("workflow reaches recursion depth {depth}; Fleet Workflow limit is {max_depth}")]
1673 RecursionTooDeep { depth: usize, max_depth: usize },
1674 #[error("expand node `{node}` must declare max_children before Fleet launch")]
1675 UnboundedExpand { node: String },
1676 #[error("expand node `{node}` must include a template before Fleet launch")]
1677 MissingExpandTemplate { node: String },
1678 #[error("loop_until node `{node}` must declare max_iterations before Fleet launch")]
1679 UnboundedLoop { node: String },
1680}
1681
1682fn estimate_fleet_shape(
1683 nodes: &[WorkflowNode],
1684) -> Result<WorkflowFleetShape, WorkflowFleetLimitError> {
1685 estimate_fleet_shape_at_depth(nodes, 1)
1686}
1687
1688fn estimate_fleet_shape_at_depth(
1689 nodes: &[WorkflowNode],
1690 depth: usize,
1691) -> Result<WorkflowFleetShape, WorkflowFleetLimitError> {
1692 nodes
1693 .iter()
1694 .try_fold(WorkflowFleetShape::default(), |shape, node| {
1695 Ok(shape.add(estimate_node_fleet_shape(node, depth)?))
1696 })
1697}
1698
1699fn estimate_node_fleet_shape(
1700 node: &WorkflowNode,
1701 depth: usize,
1702) -> Result<WorkflowFleetShape, WorkflowFleetLimitError> {
1703 match node {
1704 WorkflowNode::Leaf(_) => Ok(WorkflowFleetShape {
1705 total_agents: 1,
1706 max_depth: depth,
1707 }),
1708 WorkflowNode::BranchSet(spec) => estimate_fleet_shape_at_depth(&spec.children, depth + 1),
1709 WorkflowNode::Sequence(spec) => estimate_fleet_shape_at_depth(&spec.children, depth),
1710 WorkflowNode::Reduce(_) | WorkflowNode::TeacherReview(_) => Ok(WorkflowFleetShape {
1711 total_agents: 0,
1712 max_depth: 0,
1713 }),
1714 WorkflowNode::LoopUntil(spec) => {
1715 let iterations =
1716 spec.max_iterations
1717 .ok_or_else(|| WorkflowFleetLimitError::UnboundedLoop {
1718 node: spec.id.clone(),
1719 })? as usize;
1720 Ok(estimate_fleet_shape_at_depth(&spec.children, depth)?.repeat(iterations.max(1)))
1721 }
1722 WorkflowNode::Cond(spec) => Ok(estimate_fleet_shape_at_depth(&spec.then_nodes, depth)?
1723 .add(estimate_fleet_shape_at_depth(&spec.else_nodes, depth)?)),
1724 WorkflowNode::Expand(spec) => {
1725 let max_children =
1726 spec.max_children
1727 .ok_or_else(|| WorkflowFleetLimitError::UnboundedExpand {
1728 node: spec.id.clone(),
1729 })?;
1730 let template = spec.template.as_deref().ok_or_else(|| {
1731 WorkflowFleetLimitError::MissingExpandTemplate {
1732 node: spec.id.clone(),
1733 }
1734 })?;
1735 validate_workflow_node_shapes(std::slice::from_ref(template))
1736 .map_err(|source| WorkflowFleetLimitError::InvalidWorkflow { source })?;
1737 Ok(estimate_node_fleet_shape(template, depth)?.repeat(max_children))
1738 }
1739 }
1740}
1741
1742fn default_frontier_limit() -> usize {
1743 8
1744}
1745
1746fn node_id(node: &WorkflowNode) -> String {
1747 match node {
1748 WorkflowNode::BranchSet(spec) => spec.id.clone(),
1749 WorkflowNode::Leaf(spec) => spec.id.clone(),
1750 WorkflowNode::Sequence(spec) => spec.id.clone(),
1751 WorkflowNode::Reduce(spec) => spec.id.clone(),
1752 WorkflowNode::TeacherReview(spec) => spec.id.clone(),
1753 WorkflowNode::LoopUntil(spec) => spec.id.clone(),
1754 WorkflowNode::Cond(spec) => spec.id.clone(),
1755 WorkflowNode::Expand(spec) => spec.id.clone(),
1756 }
1757}
1758
1759pub(crate) fn validate_workflow_nodes(
1760 nodes: &[WorkflowNode],
1761) -> Result<(), WorkflowExecutionError> {
1762 let mut seen = BTreeSet::new();
1763 validate_workflow_nodes_inner(nodes, &mut seen)?;
1764 validate_workflow_references(nodes, &seen)
1765}
1766
1767pub(crate) fn validate_workflow_node_shapes(
1768 nodes: &[WorkflowNode],
1769) -> Result<(), WorkflowExecutionError> {
1770 let mut seen = BTreeSet::new();
1771 validate_workflow_nodes_inner(nodes, &mut seen)
1772}
1773
1774fn validate_workflow_nodes_inner(
1775 nodes: &[WorkflowNode],
1776 seen: &mut BTreeSet<String>,
1777) -> Result<(), WorkflowExecutionError> {
1778 for node in nodes {
1779 let id = node_id(node);
1780 let kind = control_kind_name(node);
1781 if id.trim().is_empty() {
1782 return Err(WorkflowExecutionError::EmptyNodeId { kind });
1783 }
1784 if !seen.insert(id.clone()) {
1785 return Err(WorkflowExecutionError::DuplicateNodeId { node: id });
1786 }
1787 match node {
1788 WorkflowNode::BranchSet(spec) => validate_workflow_nodes_inner(&spec.children, seen)?,
1789 WorkflowNode::Leaf(spec) => {
1790 if spec.prompt.trim().is_empty() {
1791 return Err(WorkflowExecutionError::EmptyLeafPrompt {
1792 leaf: spec.id.clone(),
1793 });
1794 }
1795 if let Some(role) = spec.role.as_deref() {
1796 validate_leaf_role(&spec.id, role)?;
1797 }
1798 if let Some(profile) = spec.profile.as_deref() {
1799 validate_leaf_profile(&spec.id, profile)?;
1800 }
1801 }
1802 WorkflowNode::Sequence(spec) => validate_workflow_nodes_inner(&spec.children, seen)?,
1803 WorkflowNode::LoopUntil(spec) => validate_workflow_nodes_inner(&spec.children, seen)?,
1804 WorkflowNode::Cond(spec) => {
1805 validate_workflow_nodes_inner(&spec.then_nodes, seen)?;
1806 validate_workflow_nodes_inner(&spec.else_nodes, seen)?;
1807 }
1808 WorkflowNode::Reduce(_) | WorkflowNode::TeacherReview(_) | WorkflowNode::Expand(_) => {}
1809 }
1810 }
1811 Ok(())
1812}
1813
1814fn validate_workflow_references(
1815 nodes: &[WorkflowNode],
1816 known_ids: &BTreeSet<String>,
1817) -> Result<(), WorkflowExecutionError> {
1818 for node in nodes {
1819 match node {
1820 WorkflowNode::BranchSet(spec) => {
1821 validate_workflow_references(&spec.children, known_ids)?;
1822 }
1823 WorkflowNode::Leaf(spec) => {
1824 validate_known_references(
1825 spec.id.as_str(),
1826 "depends_on_results",
1827 &spec.depends_on_results,
1828 known_ids,
1829 )?;
1830 }
1831 WorkflowNode::Sequence(spec) => {
1832 validate_workflow_references(&spec.children, known_ids)?;
1833 }
1834 WorkflowNode::Reduce(spec) => {
1835 validate_known_references(spec.id.as_str(), "inputs", &spec.inputs, known_ids)?;
1836 }
1837 WorkflowNode::TeacherReview(spec) => {
1838 validate_known_references(
1839 spec.id.as_str(),
1840 "candidates",
1841 &spec.candidates,
1842 known_ids,
1843 )?;
1844 }
1845 WorkflowNode::LoopUntil(spec) => {
1846 validate_workflow_references(&spec.children, known_ids)?;
1847 }
1848 WorkflowNode::Cond(spec) => {
1849 validate_workflow_references(&spec.then_nodes, known_ids)?;
1850 validate_workflow_references(&spec.else_nodes, known_ids)?;
1851 }
1852 WorkflowNode::Expand(_) => {}
1853 }
1854 }
1855 Ok(())
1856}
1857
1858fn validate_leaf_profile(leaf: &str, profile: &str) -> Result<(), WorkflowExecutionError> {
1861 let invalid = profile.is_empty()
1862 || profile
1863 .chars()
1864 .any(|ch| ch.is_whitespace() || matches!(ch, '"' | '\'' | '`' | '='));
1865 if invalid {
1866 return Err(WorkflowExecutionError::InvalidLeafProfile {
1867 leaf: leaf.to_string(),
1868 profile: profile.to_string(),
1869 });
1870 }
1871 Ok(())
1872}
1873
1874fn validate_leaf_role(leaf: &str, role: &str) -> Result<(), WorkflowExecutionError> {
1875 if validate_role_token(role).is_err() {
1876 return Err(WorkflowExecutionError::InvalidLeafRole {
1877 leaf: leaf.to_string(),
1878 role: role.to_string(),
1879 });
1880 }
1881 Ok(())
1882}
1883
1884fn validate_known_references(
1885 node: &str,
1886 field: &'static str,
1887 references: &[String],
1888 known_ids: &BTreeSet<String>,
1889) -> Result<(), WorkflowExecutionError> {
1890 for reference in references {
1891 if !known_ids.contains(reference) {
1892 return Err(WorkflowExecutionError::UnknownNodeReference {
1893 node: node.to_string(),
1894 field,
1895 reference: reference.clone(),
1896 });
1897 }
1898 }
1899 Ok(())
1900}
1901
1902fn control_kind_name(node: &WorkflowNode) -> &'static str {
1903 match node {
1904 WorkflowNode::BranchSet(_) => "branch_set",
1905 WorkflowNode::Leaf(_) => "leaf",
1906 WorkflowNode::Sequence(_) => "sequence",
1907 WorkflowNode::Reduce(_) => "reduce",
1908 WorkflowNode::TeacherReview(_) => "teacher_review",
1909 WorkflowNode::LoopUntil(_) => "loop_until",
1910 WorkflowNode::Cond(_) => "cond",
1911 WorkflowNode::Expand(_) => "expand",
1912 }
1913}
1914
1915#[derive(Debug, Clone, PartialEq, Eq, Error)]
1916pub enum WorkflowValidationError {
1917 #[error("{field} must not be empty")]
1918 EmptyField { field: &'static str },
1919 #[error("workflow must contain at least one phase")]
1920 EmptyWorkflow,
1921 #[error("phase `{phase}` must contain at least one task")]
1922 EmptyPhase { phase: String },
1923 #[error("max_concurrent must be between 1 and 20, got {value}")]
1924 InvalidMaxConcurrent { value: u8 },
1925 #[error("duplicate workflow phase `{phase}`")]
1926 DuplicatePhase { phase: String },
1927 #[error("duplicate workflow task `{task}`")]
1928 DuplicateTask { task: String },
1929 #[error("phase `{phase}` has invalid dependency `{dependency}`")]
1930 InvalidPhaseDependency { phase: String, dependency: String },
1931 #[error("phase dependency cycle includes `{phase}`")]
1932 PhaseDependencyCycle { phase: String },
1933 #[error("task `{task}` has invalid result dependency `{dependency}`")]
1934 InvalidTaskResultDependency { task: String, dependency: String },
1935 #[error(
1936 "task `{task}` depends on result `{dependency}` from unavailable phase `{dependency_phase}` while running in `{task_phase}`"
1937 )]
1938 UnavailableTaskResultDependency {
1939 task: String,
1940 dependency: String,
1941 dependency_phase: String,
1942 task_phase: String,
1943 },
1944 #[error("parallel read-write task `{task}` must declare a file_scope")]
1945 MissingParallelWriteScope { task: String },
1946 #[error("parallel read-write tasks `{left}` and `{right}` have overlapping file scopes")]
1947 OverlappingParallelWriteScope { left: String, right: String },
1948}
1949
1950fn default_max_concurrent() -> u8 {
1951 4
1952}
1953
1954fn validate_non_empty(field: &'static str, value: &str) -> Result<(), WorkflowValidationError> {
1955 if value.trim().is_empty() {
1956 return Err(WorkflowValidationError::EmptyField { field });
1957 }
1958 Ok(())
1959}
1960
1961fn ordered_phases(
1962 config: &WorkflowConfig,
1963 phase_indices: &BTreeMap<String, usize>,
1964) -> Result<Vec<String>, WorkflowValidationError> {
1965 let mut visiting = BTreeSet::new();
1966 let mut visited = BTreeSet::new();
1967 let mut ordered = Vec::with_capacity(config.phases.len());
1968
1969 for phase in &config.phases {
1970 visit_phase(
1971 &phase.name,
1972 config,
1973 phase_indices,
1974 &mut visiting,
1975 &mut visited,
1976 &mut ordered,
1977 )?;
1978 }
1979
1980 Ok(ordered)
1981}
1982
1983fn visit_phase(
1984 phase_name: &str,
1985 config: &WorkflowConfig,
1986 phase_indices: &BTreeMap<String, usize>,
1987 visiting: &mut BTreeSet<String>,
1988 visited: &mut BTreeSet<String>,
1989 ordered: &mut Vec<String>,
1990) -> Result<(), WorkflowValidationError> {
1991 if visited.contains(phase_name) {
1992 return Ok(());
1993 }
1994 if !visiting.insert(phase_name.to_string()) {
1995 return Err(WorkflowValidationError::PhaseDependencyCycle {
1996 phase: phase_name.to_string(),
1997 });
1998 }
1999
2000 let phase = &config.phases[phase_indices[phase_name]];
2001 for dependency in &phase.depends_on {
2002 visit_phase(
2003 dependency,
2004 config,
2005 phase_indices,
2006 visiting,
2007 visited,
2008 ordered,
2009 )?;
2010 }
2011
2012 visiting.remove(phase_name);
2013 visited.insert(phase_name.to_string());
2014 ordered.push(phase_name.to_string());
2015 Ok(())
2016}
2017
2018fn validate_parallel_write_scope(phase: &Phase) -> Result<(), WorkflowValidationError> {
2019 if !phase.parallel {
2020 return Ok(());
2021 }
2022
2023 let write_tasks: Vec<_> = phase
2024 .tasks
2025 .iter()
2026 .filter(|task| task.mode == TaskMode::ReadWrite)
2027 .collect();
2028
2029 for task in &write_tasks {
2030 if task.file_scope.is_empty() {
2031 return Err(WorkflowValidationError::MissingParallelWriteScope {
2032 task: task.id.clone(),
2033 });
2034 }
2035 }
2036
2037 for (left_index, left) in write_tasks.iter().enumerate() {
2038 for right in write_tasks.iter().skip(left_index + 1) {
2039 if scopes_overlap(&left.file_scope, &right.file_scope) {
2040 return Err(WorkflowValidationError::OverlappingParallelWriteScope {
2041 left: left.id.clone(),
2042 right: right.id.clone(),
2043 });
2044 }
2045 }
2046 }
2047
2048 Ok(())
2049}
2050
2051pub fn scopes_overlap(left: &[String], right: &[String]) -> bool {
2052 left.iter().any(|left_scope| {
2053 right
2054 .iter()
2055 .any(|right_scope| scope_overlaps(left_scope, right_scope))
2056 })
2057}
2058
2059fn scope_overlaps(left: &str, right: &str) -> bool {
2060 let left = normalize_file_scope_root(left);
2061 let right = normalize_file_scope_root(right);
2062
2063 if left == right || left == "." || right == "." {
2064 return true;
2065 }
2066
2067 if left.contains('*') || right.contains('*') {
2068 return glob_prefix(&left) == glob_prefix(&right);
2069 }
2070
2071 let left_path = Path::new(&left);
2072 let right_path = Path::new(&right);
2073 left_path.starts_with(right_path) || right_path.starts_with(left_path)
2074}
2075
2076#[must_use]
2079pub fn normalize_file_scope_root(scope: &str) -> String {
2080 let trimmed = scope.trim().trim_start_matches("./").trim_end_matches('/');
2081 trimmed
2082 .strip_suffix("/**")
2083 .or_else(|| trimmed.strip_suffix("/*"))
2084 .unwrap_or(trimmed)
2085 .to_string()
2086}
2087
2088fn glob_prefix(scope: &str) -> String {
2089 scope
2090 .split('*')
2091 .next()
2092 .unwrap_or(scope)
2093 .trim_end_matches('/')
2094 .to_string()
2095}
2096
2097#[cfg(test)]
2098mod tests {
2099 use super::*;
2100
2101 fn task(id: &str) -> Task {
2102 Task {
2103 id: id.to_string(),
2104 prompt: format!("run {id}"),
2105 agent_type: AgentType::General,
2106 mode: TaskMode::ReadOnly,
2107 isolation: IsolationMode::Shared,
2108 file_scope: Vec::new(),
2109 depends_on_results: Vec::new(),
2110 max_steps: None,
2111 timeout_secs: None,
2112 }
2113 }
2114
2115 fn config(phases: Vec<Phase>) -> WorkflowConfig {
2116 WorkflowConfig {
2117 goal: "cache-change".to_string(),
2118 max_concurrent: 4,
2119 description: None,
2120 phases,
2121 }
2122 }
2123
2124 fn phase(name: &str, depends_on: &[&str], tasks: Vec<Task>) -> Phase {
2125 Phase {
2126 name: name.to_string(),
2127 description: None,
2128 depends_on: depends_on.iter().map(|value| value.to_string()).collect(),
2129 parallel: false,
2130 on_failure: FailurePolicy::SkipContinue,
2131 tasks,
2132 }
2133 }
2134
2135 fn leaf_node(id: &str) -> WorkflowNode {
2136 WorkflowNode::Leaf(LeafSpec {
2137 id: id.to_string(),
2138 prompt: format!("run {id}"),
2139 agent_type: AgentType::General,
2140 role: None,
2141 profile: None,
2142 mode: TaskMode::ReadOnly,
2143 isolation: IsolationMode::Shared,
2144 file_scope: Vec::new(),
2145 depends_on_results: Vec::new(),
2146 budget: BudgetSpec::default(),
2147 permissions: PermissionSpec::default(),
2148 model_policy: ModelPolicy::default(),
2149 })
2150 }
2151
2152 fn leaf_node_with_budget(id: &str, budget: BudgetSpec) -> WorkflowNode {
2153 WorkflowNode::Leaf(LeafSpec {
2154 id: id.to_string(),
2155 prompt: format!("run {id}"),
2156 agent_type: AgentType::General,
2157 role: None,
2158 profile: None,
2159 mode: TaskMode::ReadOnly,
2160 isolation: IsolationMode::Shared,
2161 file_scope: Vec::new(),
2162 depends_on_results: Vec::new(),
2163 budget,
2164 permissions: PermissionSpec::default(),
2165 model_policy: ModelPolicy::default(),
2166 })
2167 }
2168
2169 fn invalid_leaf_node(id: &str) -> WorkflowNode {
2170 WorkflowNode::Leaf(LeafSpec {
2171 id: id.to_string(),
2172 prompt: " ".to_string(),
2173 agent_type: AgentType::General,
2174 role: None,
2175 profile: None,
2176 mode: TaskMode::ReadOnly,
2177 isolation: IsolationMode::Shared,
2178 file_scope: Vec::new(),
2179 depends_on_results: Vec::new(),
2180 budget: BudgetSpec::default(),
2181 permissions: PermissionSpec::default(),
2182 model_policy: ModelPolicy::default(),
2183 })
2184 }
2185
2186 fn workflow_spec(nodes: Vec<WorkflowNode>) -> WorkflowSpec {
2187 WorkflowSpec {
2188 id: Some("mock-workflow".to_string()),
2189 goal: "prove mock executor control flow".to_string(),
2190 description: None,
2191 budget: BudgetSpec::default(),
2192 permissions: PermissionSpec::default(),
2193 model_policy: ModelPolicy::default(),
2194 promotion_policy: PromotionPolicy::default(),
2195 gates: Vec::new(),
2196 nodes,
2197 }
2198 }
2199
2200 fn control_result<'a>(
2201 execution: &'a WorkflowExecution,
2202 node_id: &str,
2203 ) -> &'a ControlNodeResult {
2204 execution
2205 .control_node_results
2206 .iter()
2207 .find(|result| result.node_id == node_id)
2208 .expect("control node result should exist")
2209 }
2210
2211 fn candidate(
2212 branch_id: &str,
2213 status: WorkflowRunStatus,
2214 score: u32,
2215 cost: u64,
2216 diversity_key: &str,
2217 ) -> BranchCandidate {
2218 BranchCandidate {
2219 branch_id: branch_id.to_string(),
2220 status,
2221 score,
2222 cost,
2223 diversity_key: Some(diversity_key.to_string()),
2224 }
2225 }
2226
2227 #[test]
2228 fn independent_phases_preserve_declaration_order() {
2229 let workflow = config(vec![
2230 phase("discover", &[], vec![task("scan")]),
2231 phase("report", &[], vec![task("summarize")]),
2232 ]);
2233
2234 let plan = workflow.compile().expect("workflow should compile");
2235
2236 assert_eq!(
2237 plan.phase_names().collect::<Vec<_>>(),
2238 vec!["discover", "report"]
2239 );
2240 }
2241
2242 #[test]
2243 fn dependencies_override_declaration_order_deterministically() {
2244 let workflow = config(vec![
2245 phase("review", &["implement"], vec![task("review-results")]),
2246 phase("discover", &[], vec![task("scan")]),
2247 phase("implement", &["discover"], vec![task("patch")]),
2248 phase("report", &["review"], vec![task("summarize")]),
2249 ]);
2250
2251 let plan = workflow.compile().expect("workflow should compile");
2252
2253 assert_eq!(
2254 plan.phase_names().collect::<Vec<_>>(),
2255 vec!["discover", "implement", "review", "report"]
2256 );
2257 }
2258
2259 #[test]
2260 fn rejects_empty_workflow() {
2261 let err = config(Vec::new())
2262 .validate()
2263 .expect_err("empty workflow should fail");
2264
2265 assert_eq!(err, WorkflowValidationError::EmptyWorkflow);
2266 }
2267
2268 #[test]
2269 fn rejects_empty_phase() {
2270 let err = config(vec![phase("empty", &[], Vec::new())])
2271 .validate()
2272 .expect_err("empty phase should fail");
2273
2274 assert_eq!(
2275 err,
2276 WorkflowValidationError::EmptyPhase {
2277 phase: "empty".to_string()
2278 }
2279 );
2280 }
2281
2282 #[test]
2283 fn rejects_invalid_max_concurrent() {
2284 let mut workflow = config(vec![phase("discover", &[], vec![task("scan")])]);
2285 workflow.max_concurrent = 0;
2286
2287 let err = workflow
2288 .validate()
2289 .expect_err("zero concurrency should fail");
2290
2291 assert_eq!(
2292 err,
2293 WorkflowValidationError::InvalidMaxConcurrent { value: 0 }
2294 );
2295 }
2296
2297 #[test]
2298 fn rejects_duplicate_phase_names() {
2299 let err = config(vec![
2300 phase("discover", &[], vec![task("scan")]),
2301 phase("discover", &[], vec![task("scan-again")]),
2302 ])
2303 .validate()
2304 .expect_err("duplicate phase should fail");
2305
2306 assert!(matches!(
2307 err,
2308 WorkflowValidationError::DuplicatePhase { .. }
2309 ));
2310 }
2311
2312 #[test]
2313 fn rejects_duplicate_task_ids() {
2314 let err = config(vec![
2315 phase("discover", &[], vec![task("scan")]),
2316 phase("report", &[], vec![task("scan")]),
2317 ])
2318 .validate()
2319 .expect_err("duplicate task should fail");
2320
2321 assert!(matches!(err, WorkflowValidationError::DuplicateTask { .. }));
2322 }
2323
2324 #[test]
2325 fn rejects_unknown_phase_dependency() {
2326 let err = config(vec![phase("report", &["missing"], vec![task("summarize")])])
2327 .validate()
2328 .expect_err("unknown dependency should fail");
2329
2330 assert!(matches!(
2331 err,
2332 WorkflowValidationError::InvalidPhaseDependency { .. }
2333 ));
2334 }
2335
2336 #[test]
2337 fn rejects_phase_dependency_cycles() {
2338 let workflow = config(vec![
2339 phase("a", &["b"], vec![task("a-task")]),
2340 phase("b", &["a"], vec![task("b-task")]),
2341 ]);
2342
2343 let err = workflow.validate().expect_err("cycle should fail");
2344
2345 assert!(matches!(
2346 err,
2347 WorkflowValidationError::PhaseDependencyCycle { .. }
2348 ));
2349 }
2350
2351 #[test]
2352 fn rejects_task_result_dependency_from_same_parallel_phase() {
2353 let mut first = task("first");
2354 first.depends_on_results.push("second".to_string());
2355 let mut parallel = phase("parallel", &[], vec![first, task("second")]);
2356 parallel.parallel = true;
2357
2358 let err = config(vec![parallel])
2359 .validate()
2360 .expect_err("same-phase result dependency should fail");
2361
2362 assert!(matches!(
2363 err,
2364 WorkflowValidationError::UnavailableTaskResultDependency { .. }
2365 ));
2366 }
2367
2368 #[test]
2369 fn rejects_task_result_dependency_from_later_phase() {
2370 let mut summarize = task("summarize");
2371 summarize.depends_on_results.push("scan".to_string());
2372 let workflow = config(vec![
2373 phase("report", &[], vec![summarize]),
2374 phase("discover", &[], vec![task("scan")]),
2375 ]);
2376
2377 let err = workflow
2378 .validate()
2379 .expect_err("later-phase result dependency should fail");
2380
2381 assert!(matches!(
2382 err,
2383 WorkflowValidationError::UnavailableTaskResultDependency { .. }
2384 ));
2385 }
2386
2387 #[test]
2388 fn allows_task_result_dependency_from_earlier_phase() {
2389 let upstream = phase("discover", &[], vec![task("scan")]);
2390 let mut summarize = task("summarize");
2391 summarize.depends_on_results.push("scan".to_string());
2392 let downstream = phase("report", &["discover"], vec![summarize]);
2393
2394 config(vec![upstream, downstream])
2395 .validate()
2396 .expect("earlier-phase result should be available");
2397 }
2398
2399 #[test]
2400 fn rejects_parallel_read_write_without_file_scope() {
2401 let mut write = task("write");
2402 write.mode = TaskMode::ReadWrite;
2403 let mut parallel = phase("parallel", &[], vec![write]);
2404 parallel.parallel = true;
2405
2406 let err = config(vec![parallel])
2407 .validate()
2408 .expect_err("write task needs a scope");
2409
2410 assert!(matches!(
2411 err,
2412 WorkflowValidationError::MissingParallelWriteScope { .. }
2413 ));
2414 }
2415
2416 #[test]
2417 fn detects_overlapping_parallel_write_scopes_with_path_boundaries() {
2418 let mut left = task("auth");
2419 left.mode = TaskMode::ReadWrite;
2420 left.file_scope = vec!["src/auth/**".to_string()];
2421 let mut right = task("auth-login");
2422 right.mode = TaskMode::ReadWrite;
2423 right.file_scope = vec!["src/auth/login.rs".to_string()];
2424 let mut parallel = phase("parallel", &[], vec![left, right]);
2425 parallel.parallel = true;
2426
2427 let err = config(vec![parallel])
2428 .validate()
2429 .expect_err("nested scopes should overlap");
2430
2431 assert!(matches!(
2432 err,
2433 WorkflowValidationError::OverlappingParallelWriteScope { .. }
2434 ));
2435 }
2436
2437 #[test]
2438 fn does_not_confuse_path_prefixes_for_overlapping_scopes() {
2439 let mut left = task("auth");
2440 left.mode = TaskMode::ReadWrite;
2441 left.file_scope = vec!["src/auth/**".to_string()];
2442 let mut right = task("auth-admin");
2443 right.mode = TaskMode::ReadWrite;
2444 right.file_scope = vec!["src/auth_admin/**".to_string()];
2445 let mut parallel = phase("parallel", &[], vec![left, right]);
2446 parallel.parallel = true;
2447
2448 config(vec![parallel])
2449 .validate()
2450 .expect("component boundary scopes should not overlap");
2451 }
2452
2453 #[test]
2454 fn json_roundtrip_keeps_snake_case_enum_names() {
2455 let mut task = task("patch");
2456 task.agent_type = AgentType::Implementer;
2457 task.mode = TaskMode::ReadWrite;
2458 task.isolation = IsolationMode::Worktree;
2459 task.file_scope = vec!["src/auth/**".to_string()];
2460 let mut parallel = phase("implement", &[], vec![task]);
2461 parallel.parallel = true;
2462 parallel.on_failure = FailurePolicy::Abort;
2463 let workflow = config(vec![parallel]);
2464
2465 let json = serde_json::to_string(&workflow).expect("serialize workflow");
2466
2467 assert!(json.contains("\"agent_type\":\"implementer\""));
2468 assert!(json.contains("\"mode\":\"read_write\""));
2469 assert!(json.contains("\"isolation\":\"worktree\""));
2470 assert!(json.contains("\"on_failure\":\"abort\""));
2471
2472 let parsed: WorkflowConfig = serde_json::from_str(&json).expect("parse workflow");
2473 assert_eq!(parsed, workflow);
2474 }
2475
2476 #[test]
2477 fn isolation_auto_defaults_parallel_write_to_worktree() {
2478 assert_eq!(IsolationMode::default(), IsolationMode::Auto);
2479 assert_eq!(
2480 IsolationMode::Auto.resolve(true),
2481 IsolationMode::Worktree
2482 );
2483 assert_eq!(
2484 IsolationMode::Auto.resolve(false),
2485 IsolationMode::Shared
2486 );
2487 assert_eq!(
2489 IsolationMode::Shared.resolve(true),
2490 IsolationMode::Shared
2491 );
2492 assert!(IsolationMode::Worktree.wants_worktree(false));
2493 assert!(!IsolationMode::Shared.wants_worktree(true));
2494 }
2495
2496 #[test]
2497 fn leaf_write_capable_and_worktree_defaults() {
2498 let read_only = LeafSpec {
2499 id: "ro".to_string(),
2500 prompt: "inspect".to_string(),
2501 agent_type: AgentType::Explore,
2502 role: None,
2503 profile: None,
2504 mode: TaskMode::ReadOnly,
2505 isolation: IsolationMode::Auto,
2506 file_scope: Vec::new(),
2507 depends_on_results: Vec::new(),
2508 budget: BudgetSpec::default(),
2509 permissions: PermissionSpec::default(),
2510 model_policy: ModelPolicy::default(),
2511 };
2512 assert!(!leaf_is_write_capable(&read_only));
2513 assert!(!leaf_wants_worktree(&read_only, true));
2514
2515 let mut read_only_implementer = read_only.clone();
2516 read_only_implementer.id = "ro-implementer".to_string();
2517 read_only_implementer.agent_type = AgentType::Implementer;
2518 assert!(
2519 !leaf_is_write_capable(&read_only_implementer),
2520 "role identity must not grant write authority"
2521 );
2522 assert!(
2523 !leaf_wants_worktree(&read_only_implementer, true),
2524 "parallel read-only implementers stay shared under auto isolation"
2525 );
2526
2527 let mut write = read_only.clone();
2528 write.id = "rw".to_string();
2529 write.mode = TaskMode::ReadWrite;
2530 write.agent_type = AgentType::Implementer;
2531 assert!(leaf_is_write_capable(&write));
2532 assert!(leaf_wants_worktree(&write, true));
2534 assert!(!leaf_wants_worktree(&write, false));
2536
2537 write.isolation = IsolationMode::Shared;
2538 assert!(
2539 !leaf_wants_worktree(&write, true),
2540 "explicit shared is the same-worktree override"
2541 );
2542
2543 write.isolation = IsolationMode::Worktree;
2544 assert!(leaf_wants_worktree(&write, true));
2545 assert!(leaf_wants_worktree(&write, false));
2546 }
2547
2548 #[test]
2549 fn workflow_ir_roundtrip() {
2550 let discover_leaf = LeafSpec {
2551 id: "scan-readme".to_string(),
2552 prompt: "Inspect README setup gaps".to_string(),
2553 agent_type: AgentType::Explore,
2554 role: None,
2555 profile: Some("scout".to_string()),
2556 mode: TaskMode::ReadOnly,
2557 isolation: IsolationMode::Shared,
2558 file_scope: vec!["README.md".to_string()],
2559 depends_on_results: Vec::new(),
2560 budget: BudgetSpec {
2561 max_steps: Some(8),
2562 timeout_secs: Some(300),
2563 max_parallel: None,
2564 max_tokens: None,
2565 },
2566 permissions: PermissionSpec::default(),
2567 model_policy: ModelPolicy {
2568 provider: Some("openai".to_string()),
2569 model: Some("gpt-5.4".to_string()),
2570 fallback_models: Vec::new(),
2571 },
2572 };
2573 let workflow = WorkflowSpec {
2574 id: Some("v090-readme-check".to_string()),
2575 goal: "tighten setup docs".to_string(),
2576 description: Some("metadata-only typed Workflow IR".to_string()),
2577 budget: BudgetSpec {
2578 max_steps: Some(30),
2579 timeout_secs: Some(1_800),
2580 max_parallel: Some(2),
2581 max_tokens: None,
2582 },
2583 permissions: PermissionSpec {
2584 allow_write: false,
2585 allow_network: false,
2586 deny_all_tools: false,
2587 allowed_tools: vec!["rg".to_string()],
2588 file_scope: vec!["README.md".to_string()],
2589 },
2590 model_policy: ModelPolicy {
2591 provider: Some("openai".to_string()),
2592 model: Some("gpt-5.4".to_string()),
2593 fallback_models: vec!["gpt-5.4-mini".to_string()],
2594 },
2595 promotion_policy: PromotionPolicy {
2596 strategy: PromotionStrategy::TeacherSelected,
2597 require_teacher_review: true,
2598 min_successful_branches: Some(1),
2599 promotion_gate: PromotionGate::default(),
2600 },
2601 gates: Vec::new(),
2602 nodes: vec![
2603 WorkflowNode::BranchSet(BranchSpec {
2604 id: "discover".to_string(),
2605 description: Some("parallel doc inspection".to_string()),
2606 parallel: true,
2607 budget: BudgetSpec {
2608 max_steps: Some(12),
2609 timeout_secs: Some(600),
2610 max_parallel: Some(2),
2611 max_tokens: None,
2612 },
2613 permissions: PermissionSpec::default(),
2614 model_policy: ModelPolicy::default(),
2615 children: vec![WorkflowNode::Leaf(discover_leaf)],
2616 }),
2617 WorkflowNode::Sequence(SequenceSpec {
2618 id: "review-and-reduce".to_string(),
2619 children: vec![
2620 WorkflowNode::TeacherReview(TeacherReviewSpec {
2621 id: "select-best".to_string(),
2622 candidates: vec!["scan-readme".to_string()],
2623 promotion_policy: PromotionPolicy {
2624 strategy: PromotionStrategy::BestScore,
2625 require_teacher_review: true,
2626 min_successful_branches: Some(1),
2627 promotion_gate: PromotionGate::default(),
2628 },
2629 }),
2630 WorkflowNode::Reduce(ReduceSpec {
2631 id: "summarize".to_string(),
2632 inputs: vec!["scan-readme".to_string()],
2633 prompt: "Summarize the smallest safe patch".to_string(),
2634 model_policy: ModelPolicy::default(),
2635 }),
2636 ],
2637 }),
2638 WorkflowNode::Cond(CondSpec {
2639 id: "maybe-expand".to_string(),
2640 condition: "summary identifies multiple independent gaps".to_string(),
2641 then_nodes: vec![WorkflowNode::Expand(ExpandSpec {
2642 id: "split-followups".to_string(),
2643 source: "summarize".to_string(),
2644 max_children: None,
2645 template: Some(Box::new(WorkflowNode::Leaf(LeafSpec {
2646 id: "followup-template".to_string(),
2647 prompt: "Patch one independent gap".to_string(),
2648 agent_type: AgentType::Implementer,
2649 role: None,
2650 profile: None,
2651 mode: TaskMode::ReadWrite,
2652 isolation: IsolationMode::Worktree,
2653 file_scope: vec!["README.md".to_string()],
2654 depends_on_results: Vec::new(),
2655 budget: BudgetSpec::default(),
2656 permissions: PermissionSpec {
2657 allow_write: true,
2658 allow_network: false,
2659 deny_all_tools: false,
2660 allowed_tools: Vec::new(),
2661 file_scope: vec!["README.md".to_string()],
2662 },
2663 model_policy: ModelPolicy::default(),
2664 }))),
2665 })],
2666 else_nodes: vec![WorkflowNode::LoopUntil(LoopUntilSpec {
2667 id: "verify-once".to_string(),
2668 condition: "local verification passes".to_string(),
2669 max_iterations: Some(1),
2670 children: Vec::new(),
2671 })],
2672 }),
2673 ],
2674 };
2675
2676 let json = serde_json::to_string_pretty(&workflow).expect("serialize workflow ir");
2677
2678 assert!(json.contains("\"kind\": \"branch_set\""));
2679 assert!(json.contains("\"strategy\": \"teacher_selected\""));
2680 assert!(json.contains("\"profile\": \"scout\""));
2681 let parsed: WorkflowSpec = serde_json::from_str(&json).expect("parse workflow ir");
2682 assert_eq!(parsed, workflow);
2683
2684 let minimal: WorkflowSpec = serde_json::from_str(r#"{"goal":"ship v0.9","nodes":[]}"#)
2685 .expect("parse minimal workflow ir");
2686 assert_eq!(minimal.budget, BudgetSpec::default());
2687 assert_eq!(minimal.permissions, PermissionSpec::default());
2688 assert_eq!(minimal.model_policy, ModelPolicy::default());
2689
2690 let legacy_leaf: LeafSpec = serde_json::from_str(r#"{"id":"scan","prompt":"scan safely"}"#)
2692 .expect("parse pre-profile leaf ir");
2693 assert_eq!(legacy_leaf.profile, None);
2694 let legacy_json = serde_json::to_string(&legacy_leaf).expect("serialize legacy leaf");
2695 assert!(!legacy_json.contains("profile"));
2696 }
2697
2698 #[test]
2699 fn fleet_validation_accepts_one_hundred_agents_and_variable_models() {
2700 let nodes = (0..DEFAULT_FLEET_WORKFLOW_MAX_AGENTS)
2701 .map(|index| {
2702 let mut leaf = match leaf_node(&format!("agent-{index}")) {
2703 WorkflowNode::Leaf(leaf) => leaf,
2704 _ => unreachable!("leaf helper returns a leaf"),
2705 };
2706 leaf.model_policy = if index == 0 {
2707 ModelPolicy {
2708 provider: Some("deepseek".to_string()),
2709 model: Some("deepseek-v4-pro".to_string()),
2710 fallback_models: Vec::new(),
2711 }
2712 } else {
2713 ModelPolicy {
2714 provider: Some("deepseek".to_string()),
2715 model: Some("deepseek-v4-flash".to_string()),
2716 fallback_models: Vec::new(),
2717 }
2718 };
2719 WorkflowNode::Leaf(leaf)
2720 })
2721 .collect();
2722 let workflow = workflow_spec(nodes);
2723
2724 let shape = workflow
2725 .validate_for_fleet()
2726 .expect("one hundred agents should fit the Fleet Workflow limit");
2727
2728 assert_eq!(shape.total_agents, DEFAULT_FLEET_WORKFLOW_MAX_AGENTS);
2729 assert_eq!(shape.max_depth, 1);
2730 }
2731
2732 #[test]
2733 fn fleet_validation_rejects_more_than_one_hundred_agents() {
2734 let nodes = (0..=DEFAULT_FLEET_WORKFLOW_MAX_AGENTS)
2735 .map(|index| leaf_node(&format!("agent-{index}")))
2736 .collect();
2737 let workflow = workflow_spec(nodes);
2738
2739 let err = workflow
2740 .validate_for_fleet()
2741 .expect_err("agent population should be bounded before Fleet launch");
2742
2743 assert_eq!(
2744 err,
2745 WorkflowFleetLimitError::TooManyAgents {
2746 total_agents: DEFAULT_FLEET_WORKFLOW_MAX_AGENTS + 1,
2747 max_total_agents: DEFAULT_FLEET_WORKFLOW_MAX_AGENTS,
2748 }
2749 );
2750 }
2751
2752 #[test]
2753 fn fleet_validation_rejects_depth_beyond_five() {
2754 let mut node = leaf_node("deep-leaf");
2755 for depth in (0..DEFAULT_FLEET_WORKFLOW_MAX_DEPTH).rev() {
2756 node = WorkflowNode::BranchSet(BranchSpec {
2757 id: format!("ring-{depth}"),
2758 description: None,
2759 parallel: true,
2760 budget: BudgetSpec::default(),
2761 permissions: PermissionSpec::default(),
2762 model_policy: ModelPolicy::default(),
2763 children: vec![node],
2764 });
2765 }
2766 let workflow = workflow_spec(vec![node]);
2767
2768 let err = workflow
2769 .validate_for_fleet()
2770 .expect_err("sixth agent ring should be rejected");
2771
2772 assert_eq!(
2773 err,
2774 WorkflowFleetLimitError::RecursionTooDeep {
2775 depth: DEFAULT_FLEET_WORKFLOW_MAX_DEPTH + 1,
2776 max_depth: DEFAULT_FLEET_WORKFLOW_MAX_DEPTH,
2777 }
2778 );
2779 }
2780
2781 #[test]
2782 fn fleet_validation_counts_loop_and_expand_fanout_conservatively() {
2783 let workflow = workflow_spec(vec![
2784 WorkflowNode::LoopUntil(LoopUntilSpec {
2785 id: "retry-ring".to_string(),
2786 condition: "verifier passes".to_string(),
2787 max_iterations: Some(3),
2788 children: vec![leaf_node("retry-worker")],
2789 }),
2790 WorkflowNode::Expand(ExpandSpec {
2791 id: "split".to_string(),
2792 source: "retry-ring".to_string(),
2793 max_children: Some(4),
2794 template: Some(Box::new(leaf_node("split-template"))),
2795 }),
2796 ]);
2797
2798 let shape = workflow
2799 .validate_for_fleet()
2800 .expect("bounded loop and expand should validate");
2801
2802 assert_eq!(shape.total_agents, 7);
2803 assert_eq!(shape.max_depth, 1);
2804 }
2805
2806 #[test]
2807 fn fleet_validation_rejects_unbounded_loop_or_expand_before_launch() {
2808 let workflow = workflow_spec(vec![
2809 WorkflowNode::LoopUntil(LoopUntilSpec {
2810 id: "retry-ring".to_string(),
2811 condition: "verifier passes".to_string(),
2812 max_iterations: None,
2813 children: vec![leaf_node("retry-worker")],
2814 }),
2815 WorkflowNode::Expand(ExpandSpec {
2816 id: "split".to_string(),
2817 source: "retry-ring".to_string(),
2818 max_children: Some(4),
2819 template: Some(Box::new(leaf_node("split-template"))),
2820 }),
2821 ]);
2822
2823 assert!(matches!(
2824 workflow.validate_for_fleet(),
2825 Err(WorkflowFleetLimitError::UnboundedLoop { node }) if node == "retry-ring"
2826 ));
2827
2828 let workflow = workflow_spec(vec![WorkflowNode::Expand(ExpandSpec {
2829 id: "split".to_string(),
2830 source: "retry-ring".to_string(),
2831 max_children: None,
2832 template: Some(Box::new(leaf_node("split-template"))),
2833 })]);
2834
2835 assert!(matches!(
2836 workflow.validate_for_fleet(),
2837 Err(WorkflowFleetLimitError::UnboundedExpand { node }) if node == "split"
2838 ));
2839 }
2840
2841 #[test]
2842 fn branch_result_serialization() {
2843 let result = BranchResult {
2844 branch_id: "discover".to_string(),
2845 task_id: "scan".to_string(),
2846 status: WorkflowRunStatus::Succeeded,
2847 usage: WorkflowUsage {
2848 input_tokens: 100,
2849 output_tokens: 25,
2850 cost_microusd: 42,
2851 },
2852 memo_usage: WorkflowMemoUsage::default(),
2853 artifacts: vec!["trace://branches/discover".to_string()],
2854 notes: Some("validated prompt surfaces".to_string()),
2855 };
2856
2857 let json = serde_json::to_string(&result).expect("serialize branch result");
2858
2859 assert!(json.contains("\"status\":\"succeeded\""));
2860 assert!(json.contains("\"cost_microusd\":42"));
2861 let parsed: BranchResult = serde_json::from_str(&json).expect("parse branch result");
2862 assert_eq!(parsed, result);
2863
2864 let minimal: BranchResult =
2865 serde_json::from_str(r#"{"branch_id":"discover","task_id":"scan","status":"pending"}"#)
2866 .expect("parse minimal branch result");
2867 assert_eq!(minimal.usage, WorkflowUsage::default());
2868 assert_eq!(minimal.memo_usage, WorkflowMemoUsage::default());
2869 assert!(minimal.artifacts.is_empty());
2870 assert_eq!(minimal.notes, None);
2871 }
2872
2873 #[test]
2874 fn leaf_result_serialization() {
2875 let result = LeafResult {
2876 leaf_id: "scan-readme".to_string(),
2877 task_id: "scan".to_string(),
2878 role: None,
2879 profile: Some("reviewer".to_string()),
2880 status: WorkflowRunStatus::Failed,
2881 usage: WorkflowUsage {
2882 input_tokens: 11,
2883 output_tokens: 7,
2884 cost_microusd: 3,
2885 },
2886 memo_usage: WorkflowMemoUsage {
2887 armh_hits: 1,
2888 armh_misses: 0,
2889 armh_saved_estimated_tokens: 128,
2890 provider_prompt_cache_hits: 2,
2891 provider_prompt_cache_misses: 1,
2892 },
2893 output: Some("README needs clearer setup steps".to_string()),
2894 artifacts: vec!["trace://leaves/scan-readme".to_string()],
2895 schema_error: None,
2896 };
2897
2898 let json = serde_json::to_string(&result).expect("serialize leaf result");
2899
2900 assert!(json.contains("\"status\":\"failed\""));
2901 assert!(json.contains("\"input_tokens\":11"));
2902 assert!(json.contains("\"armh_saved_estimated_tokens\":128"));
2903 assert!(json.contains("\"profile\":\"reviewer\""));
2904 let parsed: LeafResult = serde_json::from_str(&json).expect("parse leaf result");
2905 assert_eq!(parsed, result);
2906
2907 let minimal: LeafResult = serde_json::from_str(
2908 r#"{"leaf_id":"scan-readme","task_id":"scan","status":"pending"}"#,
2909 )
2910 .expect("parse minimal leaf result");
2911 assert_eq!(minimal.profile, None);
2912 assert_eq!(minimal.usage, WorkflowUsage::default());
2913 assert_eq!(minimal.memo_usage, WorkflowMemoUsage::default());
2914 assert_eq!(minimal.output, None);
2915 assert!(minimal.artifacts.is_empty());
2916 }
2917
2918 #[test]
2919 fn control_node_result_serialization() {
2920 let result = ControlNodeResult {
2921 node_id: "select-fix".to_string(),
2922 kind: ControlNodeKind::TeacherReview,
2923 status: WorkflowRunStatus::Running,
2924 selected_children: vec!["branch-a".to_string(), "branch-c".to_string()],
2925 summary: Some("teacher review is waiting on verifier evidence".to_string()),
2926 };
2927
2928 let json = serde_json::to_string(&result).expect("serialize control node result");
2929
2930 assert!(json.contains("\"kind\":\"teacher_review\""));
2931 assert!(json.contains("\"status\":\"running\""));
2932 let parsed: ControlNodeResult =
2933 serde_json::from_str(&json).expect("parse control node result");
2934 assert_eq!(parsed, result);
2935
2936 let minimal: ControlNodeResult = serde_json::from_str(
2937 r#"{"node_id":"select-fix","kind":"branch_set","status":"pending"}"#,
2938 )
2939 .expect("parse minimal control node result");
2940 assert!(minimal.selected_children.is_empty());
2941 assert_eq!(minimal.summary, None);
2942 }
2943
2944 #[test]
2945 fn run_mock_three_branch_workflow() {
2946 let workflow = workflow_spec(vec![WorkflowNode::BranchSet(BranchSpec {
2947 id: "discover".to_string(),
2948 description: None,
2949 parallel: true,
2950 budget: BudgetSpec::default(),
2951 permissions: PermissionSpec::default(),
2952 model_policy: ModelPolicy::default(),
2953 children: vec![
2954 leaf_node("scan-readme"),
2955 leaf_node("scan-config"),
2956 leaf_node("scan-tests"),
2957 ],
2958 })]);
2959
2960 let mut executor = MockWorkflowExecutor::new();
2961 let execution = executor.run(&workflow).expect("mock workflow should run");
2962
2963 assert_eq!(execution.status, WorkflowRunStatus::Succeeded);
2964 assert_eq!(
2965 execution
2966 .leaf_results
2967 .iter()
2968 .map(|result| result.leaf_id.as_str())
2969 .collect::<Vec<_>>(),
2970 vec!["scan-readme", "scan-config", "scan-tests"]
2971 );
2972 assert_eq!(execution.branch_results.len(), 1);
2973 assert_eq!(execution.branch_results[0].branch_id, "discover");
2974 assert_eq!(
2975 control_result(&execution, "discover").selected_children,
2976 vec!["scan-readme", "scan-config", "scan-tests"]
2977 );
2978 }
2979
2980 #[test]
2981 fn mock_executor_surfaces_leaf_profile() {
2982 let mut profiled_leaf = match leaf_node("review-change") {
2983 WorkflowNode::Leaf(leaf) => leaf,
2984 _ => unreachable!("leaf helper returns a leaf"),
2985 };
2986 profiled_leaf.profile = Some("reviewer".to_string());
2987 let workflow = workflow_spec(vec![
2988 WorkflowNode::Leaf(profiled_leaf),
2989 leaf_node("scan-readme"),
2990 ]);
2991
2992 let execution = MockWorkflowExecutor::new()
2993 .run(&workflow)
2994 .expect("mock workflow should run");
2995
2996 assert_eq!(execution.status, WorkflowRunStatus::Succeeded);
2997 assert_eq!(
2998 execution.leaf_results[0].profile.as_deref(),
2999 Some("reviewer")
3000 );
3001 assert_eq!(execution.leaf_results[1].profile, None);
3002 }
3003
3004 #[test]
3005 fn mock_executor_surfaces_leaf_role() {
3006 let mut role_leaf = match leaf_node("scout-issue") {
3007 WorkflowNode::Leaf(leaf) => leaf,
3008 _ => unreachable!("leaf helper returns a leaf"),
3009 };
3010 role_leaf.role = Some("scout".to_string());
3011 let workflow = workflow_spec(vec![WorkflowNode::Leaf(role_leaf)]);
3012
3013 let execution = MockWorkflowExecutor::new()
3014 .run(&workflow)
3015 .expect("mock workflow should run");
3016
3017 assert_eq!(execution.leaf_results[0].role.as_deref(), Some("scout"));
3018 }
3019
3020 #[test]
3021 fn leaf_role_token_rule_rejects_invalid_names() {
3022 for bad in ["", "has space", "role=scout"] {
3023 let mut leaf = match leaf_node("scan") {
3024 WorkflowNode::Leaf(leaf) => leaf,
3025 _ => unreachable!("leaf helper returns a leaf"),
3026 };
3027 leaf.role = Some(bad.to_string());
3028 let workflow = workflow_spec(vec![WorkflowNode::Leaf(leaf)]);
3029
3030 let err = MockWorkflowExecutor::new()
3031 .run(&workflow)
3032 .expect_err("invalid role token should fail validation");
3033
3034 assert!(
3035 matches!(&err, WorkflowExecutionError::InvalidLeafRole { role, .. } if role == bad),
3036 "role `{bad}` should be rejected, got {err:?}"
3037 );
3038 }
3039 }
3040
3041 #[test]
3042 fn leaf_role_roundtrips_without_required_provider_model() {
3043 let leaf = LeafSpec {
3044 id: "scout-1".to_string(),
3045 prompt: "Investigate #4090. Read-only.".to_string(),
3046 agent_type: AgentType::Explore,
3047 role: Some("scout".to_string()),
3048 profile: None,
3049 mode: TaskMode::ReadOnly,
3050 isolation: IsolationMode::Shared,
3051 file_scope: Vec::new(),
3052 depends_on_results: Vec::new(),
3053 budget: BudgetSpec::default(),
3054 permissions: PermissionSpec::default(),
3055 model_policy: ModelPolicy::default(),
3056 };
3057 let json = serde_json::to_string(&leaf).expect("serialize");
3058 assert!(json.contains("\"role\":\"scout\""));
3059 let parsed: LeafSpec = serde_json::from_str(&json).expect("parse");
3060 assert_eq!(parsed.role.as_deref(), Some("scout"));
3061 assert_eq!(parsed.model_policy.provider, None);
3063 assert_eq!(parsed.model_policy.model, None);
3064 assert_eq!(parsed.model_policy, ModelPolicy::default());
3065 }
3066
3067 #[test]
3068 fn leaf_profile_token_rule_rejects_invalid_names() {
3069 for bad in ["", "has space", "quote\"y", "role=reviewer", "back`tick"] {
3070 let mut leaf = match leaf_node("scan") {
3071 WorkflowNode::Leaf(leaf) => leaf,
3072 _ => unreachable!("leaf helper returns a leaf"),
3073 };
3074 leaf.profile = Some(bad.to_string());
3075 let workflow = workflow_spec(vec![WorkflowNode::Leaf(leaf)]);
3076
3077 let err = MockWorkflowExecutor::new()
3078 .run(&workflow)
3079 .expect_err("invalid profile token should fail validation");
3080
3081 assert!(
3082 matches!(&err, WorkflowExecutionError::InvalidLeafProfile { profile, .. } if profile == bad),
3083 "profile `{bad}` should be rejected, got {err:?}"
3084 );
3085 }
3086
3087 let mut leaf = match leaf_node("scan") {
3088 WorkflowNode::Leaf(leaf) => leaf,
3089 _ => unreachable!("leaf helper returns a leaf"),
3090 };
3091 leaf.profile = Some("reviewer".to_string());
3092 let workflow = workflow_spec(vec![WorkflowNode::Leaf(leaf)]);
3093 MockWorkflowExecutor::new()
3094 .run(&workflow)
3095 .expect("valid profile token should pass validation");
3096 }
3097
3098 #[test]
3099 fn mock_executor_aggregates_leaf_usage() {
3100 let workflow = workflow_spec(vec![WorkflowNode::BranchSet(BranchSpec {
3101 id: "discover".to_string(),
3102 description: None,
3103 parallel: true,
3104 budget: BudgetSpec::default(),
3105 permissions: PermissionSpec::default(),
3106 model_policy: ModelPolicy::default(),
3107 children: vec![leaf_node("scan-readme"), leaf_node("scan-tests")],
3108 })]);
3109
3110 let mut executor = MockWorkflowExecutor::new()
3111 .with_leaf_outcome(
3112 "scan-readme",
3113 MockLeafOutcome::succeeded("readme ok").with_usage(WorkflowUsage {
3114 input_tokens: 100,
3115 output_tokens: 25,
3116 cost_microusd: 500,
3117 }),
3118 )
3119 .with_leaf_outcome(
3120 "scan-tests",
3121 MockLeafOutcome::succeeded("tests ok").with_usage(WorkflowUsage {
3122 input_tokens: 50,
3123 output_tokens: 10,
3124 cost_microusd: 250,
3125 }),
3126 );
3127
3128 let execution = executor.run(&workflow).expect("mock workflow should run");
3129
3130 assert_eq!(
3131 execution.usage,
3132 WorkflowUsage {
3133 input_tokens: 150,
3134 output_tokens: 35,
3135 cost_microusd: 750,
3136 }
3137 );
3138 assert_eq!(execution.usage.total_tokens(), 185);
3139 assert_eq!(execution.branch_results[0].usage, execution.usage);
3140 assert_eq!(
3141 execution
3142 .leaf_results
3143 .iter()
3144 .map(|result| result.usage.cost_microusd)
3145 .collect::<Vec<_>>(),
3146 vec![500, 250]
3147 );
3148 }
3149
3150 #[test]
3151 fn mock_executor_aggregates_memo_usage() {
3152 let workflow = workflow_spec(vec![WorkflowNode::BranchSet(BranchSpec {
3153 id: "cache-branches".to_string(),
3154 description: None,
3155 parallel: true,
3156 budget: BudgetSpec::default(),
3157 permissions: PermissionSpec::default(),
3158 model_policy: ModelPolicy::default(),
3159 children: vec![leaf_node("rlm-hit"), leaf_node("rlm-miss")],
3160 })]);
3161
3162 let mut executor = MockWorkflowExecutor::new()
3163 .with_leaf_outcome(
3164 "rlm-hit",
3165 MockLeafOutcome::succeeded("memo hit").with_memo_usage(WorkflowMemoUsage {
3166 armh_hits: 1,
3167 armh_misses: 0,
3168 armh_saved_estimated_tokens: 4096,
3169 provider_prompt_cache_hits: 1,
3170 provider_prompt_cache_misses: 0,
3171 }),
3172 )
3173 .with_leaf_outcome(
3174 "rlm-miss",
3175 MockLeafOutcome::succeeded("memo miss").with_memo_usage(WorkflowMemoUsage {
3176 armh_hits: 0,
3177 armh_misses: 1,
3178 armh_saved_estimated_tokens: 0,
3179 provider_prompt_cache_hits: 0,
3180 provider_prompt_cache_misses: 1,
3181 }),
3182 );
3183
3184 let execution = executor.run(&workflow).expect("mock workflow should run");
3185
3186 assert_eq!(
3187 execution.memo_usage,
3188 WorkflowMemoUsage {
3189 armh_hits: 1,
3190 armh_misses: 1,
3191 armh_saved_estimated_tokens: 4096,
3192 provider_prompt_cache_hits: 1,
3193 provider_prompt_cache_misses: 1,
3194 }
3195 );
3196 assert_eq!(execution.branch_results[0].memo_usage, execution.memo_usage);
3197 assert_eq!(
3198 execution
3199 .leaf_results
3200 .iter()
3201 .map(|result| (result.memo_usage.armh_hits, result.memo_usage.armh_misses))
3202 .collect::<Vec<_>>(),
3203 vec![(1, 0), (0, 1)]
3204 );
3205 }
3206
3207 #[test]
3208 fn mock_executor_marks_cancelled_before_leaf() {
3209 let workflow = workflow_spec(vec![WorkflowNode::BranchSet(BranchSpec {
3210 id: "discover".to_string(),
3211 description: None,
3212 parallel: true,
3213 budget: BudgetSpec::default(),
3214 permissions: PermissionSpec::default(),
3215 model_policy: ModelPolicy::default(),
3216 children: vec![leaf_node("scan-readme"), leaf_node("scan-tests")],
3217 })]);
3218
3219 let mut executor = MockWorkflowExecutor::new().with_cancelled();
3220 let execution = executor.run(&workflow).expect("mock workflow should run");
3221
3222 assert_eq!(execution.status, WorkflowRunStatus::Cancelled);
3223 assert_eq!(execution.leaf_results.len(), 1);
3224 assert_eq!(
3225 execution.leaf_results[0].status,
3226 WorkflowRunStatus::Cancelled
3227 );
3228 assert_eq!(
3229 execution.branch_results[0].status,
3230 WorkflowRunStatus::Cancelled
3231 );
3232 assert_eq!(
3233 control_result(&execution, "discover").status,
3234 WorkflowRunStatus::Cancelled
3235 );
3236 }
3237
3238 #[test]
3239 fn mock_executor_stops_when_global_leaf_budget_is_exhausted() {
3240 let workflow = workflow_spec(vec![WorkflowNode::BranchSet(BranchSpec {
3241 id: "discover".to_string(),
3242 description: None,
3243 parallel: true,
3244 budget: BudgetSpec::default(),
3245 permissions: PermissionSpec::default(),
3246 model_policy: ModelPolicy::default(),
3247 children: vec![
3248 leaf_node("scan-readme"),
3249 leaf_node("scan-config"),
3250 leaf_node("scan-tests"),
3251 ],
3252 })]);
3253
3254 let mut executor = MockWorkflowExecutor::new().with_max_leaf_steps(1);
3255 let execution = executor.run(&workflow).expect("mock workflow should run");
3256
3257 assert_eq!(execution.status, WorkflowRunStatus::BudgetExceeded);
3258 assert_eq!(
3259 execution
3260 .leaf_results
3261 .iter()
3262 .map(|result| (result.leaf_id.as_str(), result.status))
3263 .collect::<Vec<_>>(),
3264 vec![
3265 ("scan-readme", WorkflowRunStatus::Succeeded),
3266 ("scan-config", WorkflowRunStatus::BudgetExceeded)
3267 ]
3268 );
3269 assert_eq!(
3270 execution.branch_results[0].status,
3271 WorkflowRunStatus::BudgetExceeded
3272 );
3273 }
3274
3275 #[test]
3276 fn mock_executor_honors_zero_step_leaf_budget() {
3277 let workflow = workflow_spec(vec![WorkflowNode::BranchSet(BranchSpec {
3278 id: "verify".to_string(),
3279 description: None,
3280 parallel: false,
3281 budget: BudgetSpec::default(),
3282 permissions: PermissionSpec::default(),
3283 model_policy: ModelPolicy::default(),
3284 children: vec![
3285 leaf_node_with_budget(
3286 "run-tests",
3287 BudgetSpec {
3288 max_steps: Some(0),
3289 timeout_secs: None,
3290 max_parallel: None,
3291 max_tokens: None,
3292 },
3293 ),
3294 leaf_node("summarize"),
3295 ],
3296 })]);
3297
3298 let mut executor = MockWorkflowExecutor::new();
3299 let execution = executor.run(&workflow).expect("mock workflow should run");
3300
3301 assert_eq!(execution.status, WorkflowRunStatus::BudgetExceeded);
3302 assert_eq!(execution.leaf_results.len(), 1);
3303 assert_eq!(
3304 execution.leaf_results[0].status,
3305 WorkflowRunStatus::BudgetExceeded
3306 );
3307 assert!(
3308 execution.leaf_results[0]
3309 .output
3310 .as_deref()
3311 .unwrap_or_default()
3312 .contains("budget exhausted")
3313 );
3314 }
3315
3316 #[test]
3317 fn mock_executor_stops_when_global_token_budget_is_exhausted() {
3318 let workflow = workflow_spec(vec![WorkflowNode::BranchSet(BranchSpec {
3319 id: "discover".to_string(),
3320 description: None,
3321 parallel: true,
3322 budget: BudgetSpec::default(),
3323 permissions: PermissionSpec::default(),
3324 model_policy: ModelPolicy::default(),
3325 children: vec![
3326 leaf_node("scan-readme"),
3327 leaf_node("scan-config"),
3328 leaf_node("scan-tests"),
3329 ],
3330 })]);
3331
3332 let mut executor = MockWorkflowExecutor::new()
3337 .with_max_leaf_tokens(1000)
3338 .with_leaf_outcome(
3339 "scan-readme",
3340 MockLeafOutcome::succeeded("readme done").with_usage(WorkflowUsage {
3341 input_tokens: 300,
3342 output_tokens: 300,
3343 cost_microusd: 0,
3344 }),
3345 )
3346 .with_leaf_outcome(
3347 "scan-config",
3348 MockLeafOutcome::succeeded("config done").with_usage(WorkflowUsage {
3349 input_tokens: 250,
3350 output_tokens: 250,
3351 cost_microusd: 0,
3352 }),
3353 );
3354 let execution = executor.run(&workflow).expect("mock workflow should run");
3355
3356 assert_eq!(execution.status, WorkflowRunStatus::BudgetExceeded);
3357 assert_eq!(execution.leaf_results.len(), 3);
3363 assert_eq!(
3364 execution.leaf_results[0].status,
3365 WorkflowRunStatus::Succeeded
3366 );
3367 assert_eq!(
3368 execution.leaf_results[1].status,
3369 WorkflowRunStatus::Succeeded
3370 );
3371 assert_eq!(
3372 execution.leaf_results[2].status,
3373 WorkflowRunStatus::BudgetExceeded
3374 );
3375 assert_eq!(execution.usage.total_tokens(), 1100);
3376 }
3377
3378 #[test]
3379 fn mock_executor_honors_zero_token_leaf_budget() {
3380 let workflow = workflow_spec(vec![WorkflowNode::BranchSet(BranchSpec {
3381 id: "verify".to_string(),
3382 description: None,
3383 parallel: false,
3384 budget: BudgetSpec::default(),
3385 permissions: PermissionSpec::default(),
3386 model_policy: ModelPolicy::default(),
3387 children: vec![
3388 leaf_node_with_budget(
3389 "run-tests",
3390 BudgetSpec {
3391 max_steps: None,
3392 timeout_secs: None,
3393 max_parallel: None,
3394 max_tokens: Some(0),
3395 },
3396 ),
3397 leaf_node("summarize"),
3398 ],
3399 })]);
3400
3401 let mut executor = MockWorkflowExecutor::new();
3402 let execution = executor.run(&workflow).expect("mock workflow should run");
3403
3404 assert_eq!(execution.status, WorkflowRunStatus::BudgetExceeded);
3405 assert_eq!(execution.leaf_results.len(), 1);
3406 assert_eq!(
3407 execution.leaf_results[0].status,
3408 WorkflowRunStatus::BudgetExceeded
3409 );
3410 assert!(
3411 execution.leaf_results[0]
3412 .output
3413 .as_deref()
3414 .unwrap_or_default()
3415 .contains("token budget exhausted")
3416 );
3417 }
3418
3419 #[test]
3420 fn mock_executor_honors_per_leaf_token_cap() {
3421 let workflow = workflow_spec(vec![WorkflowNode::BranchSet(BranchSpec {
3422 id: "review".to_string(),
3423 description: None,
3424 parallel: false,
3425 budget: BudgetSpec::default(),
3426 permissions: PermissionSpec::default(),
3427 model_policy: ModelPolicy::default(),
3428 children: vec![
3429 leaf_node_with_budget(
3430 "expensive-scan",
3431 BudgetSpec {
3432 max_steps: None,
3433 timeout_secs: None,
3434 max_parallel: None,
3435 max_tokens: Some(500),
3436 },
3437 ),
3438 leaf_node("summarize"),
3439 ],
3440 })]);
3441
3442 let mut executor = MockWorkflowExecutor::new().with_leaf_outcome(
3444 "expensive-scan",
3445 MockLeafOutcome::succeeded("scan done").with_usage(WorkflowUsage {
3446 input_tokens: 500,
3447 output_tokens: 300,
3448 cost_microusd: 0,
3449 }),
3450 );
3451 let execution = executor.run(&workflow).expect("mock workflow should run");
3452
3453 assert_eq!(execution.status, WorkflowRunStatus::BudgetExceeded);
3454 assert_eq!(execution.leaf_results.len(), 1);
3455 assert_eq!(
3456 execution.leaf_results[0].status,
3457 WorkflowRunStatus::BudgetExceeded
3458 );
3459 assert!(
3460 execution.leaf_results[0]
3461 .output
3462 .as_deref()
3463 .unwrap_or_default()
3464 .contains("token budget exhausted")
3465 );
3466 }
3467
3468 #[test]
3469 fn budget_spec_serializes_max_tokens() {
3470 let budget = BudgetSpec {
3471 max_steps: Some(10),
3472 timeout_secs: Some(600),
3473 max_parallel: Some(4),
3474 max_tokens: Some(50_000),
3475 };
3476 let json = serde_json::to_string(&budget).expect("serialize budget");
3477 let parsed: BudgetSpec = serde_json::from_str(&json).expect("parse budget");
3478 assert_eq!(parsed, budget);
3479 assert!(json.contains("\"max_tokens\":50000"));
3480
3481 let default_json =
3483 serde_json::to_string(&BudgetSpec::default()).expect("serialize default");
3484 let parsed_default: BudgetSpec =
3485 serde_json::from_str(&default_json).expect("parse default budget");
3486 assert_eq!(parsed_default, BudgetSpec::default());
3487 assert!(parsed_default.max_tokens.is_none());
3488 }
3489
3490 #[test]
3491 fn loop_until_stops_on_pass() {
3492 let workflow = workflow_spec(vec![WorkflowNode::LoopUntil(LoopUntilSpec {
3493 id: "verify".to_string(),
3494 condition: "verification passed".to_string(),
3495 max_iterations: Some(5),
3496 children: vec![leaf_node("run-check")],
3497 })]);
3498
3499 let mut executor =
3500 MockWorkflowExecutor::new().with_predicate_results("verify", vec![false, false, true]);
3501 let execution = executor.run(&workflow).expect("loop should run");
3502
3503 assert_eq!(execution.status, WorkflowRunStatus::Succeeded);
3504 assert_eq!(execution.leaf_results.len(), 3);
3505 assert_eq!(
3506 control_result(&execution, "verify").summary.as_deref(),
3507 Some("loop_until iterations=3")
3508 );
3509 }
3510
3511 #[test]
3512 fn loop_until_honors_max_iters() {
3513 let workflow = workflow_spec(vec![WorkflowNode::LoopUntil(LoopUntilSpec {
3514 id: "verify".to_string(),
3515 condition: "verification passed".to_string(),
3516 max_iterations: Some(2),
3517 children: vec![leaf_node("run-check")],
3518 })]);
3519
3520 let mut executor =
3521 MockWorkflowExecutor::new().with_predicate_results("verify", vec![false, false, true]);
3522 let execution = executor.run(&workflow).expect("loop should run");
3523
3524 assert_eq!(execution.status, WorkflowRunStatus::Failed);
3525 assert_eq!(execution.leaf_results.len(), 2);
3526 assert_eq!(
3527 control_result(&execution, "verify").summary.as_deref(),
3528 Some("loop_until iterations=2")
3529 );
3530 }
3531
3532 #[test]
3533 fn cond_uses_logged_predicate_result() {
3534 let workflow = workflow_spec(vec![WorkflowNode::Cond(CondSpec {
3535 id: "should-fix".to_string(),
3536 condition: "finding requires a patch".to_string(),
3537 then_nodes: vec![leaf_node("patch")],
3538 else_nodes: vec![leaf_node("report-only")],
3539 })]);
3540
3541 let mut executor =
3542 MockWorkflowExecutor::new().with_predicate_results("should-fix", vec![true]);
3543 let execution = executor.run(&workflow).expect("cond should run");
3544
3545 assert_eq!(
3546 execution
3547 .leaf_results
3548 .iter()
3549 .map(|result| result.leaf_id.as_str())
3550 .collect::<Vec<_>>(),
3551 vec!["patch"]
3552 );
3553 assert_eq!(
3554 control_result(&execution, "should-fix").summary.as_deref(),
3555 Some("predicate_result=true")
3556 );
3557 }
3558
3559 #[test]
3560 fn expand_respects_max_children() {
3561 let workflow = workflow_spec(vec![WorkflowNode::Expand(ExpandSpec {
3562 id: "split".to_string(),
3563 source: "plan".to_string(),
3564 max_children: Some(2),
3565 template: None,
3566 })]);
3567
3568 let generated = vec![leaf_node("first"), leaf_node("second"), leaf_node("third")];
3569 let mut executor = MockWorkflowExecutor::new().with_generated_nodes("split", generated);
3570 let execution = executor.run(&workflow).expect("expand should run");
3571
3572 assert_eq!(
3573 execution
3574 .leaf_results
3575 .iter()
3576 .map(|result| result.leaf_id.as_str())
3577 .collect::<Vec<_>>(),
3578 vec!["first", "second"]
3579 );
3580 assert_eq!(
3581 control_result(&execution, "split").selected_children,
3582 vec!["first", "second"]
3583 );
3584 }
3585
3586 #[test]
3587 fn expand_generated_nodes_validate_before_run() {
3588 let workflow = workflow_spec(vec![WorkflowNode::Expand(ExpandSpec {
3589 id: "split".to_string(),
3590 source: "plan".to_string(),
3591 max_children: None,
3592 template: None,
3593 })]);
3594
3595 let mut executor = MockWorkflowExecutor::new()
3596 .with_generated_nodes("split", vec![invalid_leaf_node("bad")]);
3597 let err = executor
3598 .run(&workflow)
3599 .expect_err("invalid generated leaf should fail before execution");
3600
3601 assert_eq!(
3602 err,
3603 WorkflowExecutionError::EmptyLeafPrompt {
3604 leaf: "bad".to_string()
3605 }
3606 );
3607 }
3608
3609 #[test]
3610 fn workflow_spec_rejects_unknown_leaf_dependency() {
3611 let mut summarize = leaf_node("summarize");
3612 let WorkflowNode::Leaf(spec) = &mut summarize else {
3613 panic!("expected leaf");
3614 };
3615 spec.depends_on_results = vec!["missing-scan".to_string()];
3616 let workflow = workflow_spec(vec![summarize]);
3617
3618 let mut executor = MockWorkflowExecutor::new();
3619 let err = executor
3620 .run(&workflow)
3621 .expect_err("unknown leaf dependency should fail before execution");
3622
3623 assert_eq!(
3624 err,
3625 WorkflowExecutionError::UnknownNodeReference {
3626 node: "summarize".to_string(),
3627 field: "depends_on_results",
3628 reference: "missing-scan".to_string(),
3629 }
3630 );
3631 }
3632
3633 #[test]
3634 fn workflow_spec_rejects_unknown_reduce_input() {
3635 let workflow = workflow_spec(vec![
3636 leaf_node("scan"),
3637 WorkflowNode::Reduce(ReduceSpec {
3638 id: "summarize".to_string(),
3639 inputs: vec!["scan".to_string(), "missing-review".to_string()],
3640 prompt: "Summarize safe fixes".to_string(),
3641 model_policy: ModelPolicy::default(),
3642 }),
3643 ]);
3644
3645 let mut executor = MockWorkflowExecutor::new();
3646 let err = executor
3647 .run(&workflow)
3648 .expect_err("unknown reduce input should fail before execution");
3649
3650 assert_eq!(
3651 err,
3652 WorkflowExecutionError::UnknownNodeReference {
3653 node: "summarize".to_string(),
3654 field: "inputs",
3655 reference: "missing-review".to_string(),
3656 }
3657 );
3658 }
3659
3660 #[test]
3661 fn workflow_spec_rejects_unknown_teacher_candidate() {
3662 let workflow = workflow_spec(vec![
3663 leaf_node("candidate-a"),
3664 WorkflowNode::TeacherReview(TeacherReviewSpec {
3665 id: "teacher-review".to_string(),
3666 candidates: vec!["candidate-a".to_string(), "candidate-b".to_string()],
3667 promotion_policy: PromotionPolicy::default(),
3668 }),
3669 ]);
3670
3671 let mut executor = MockWorkflowExecutor::new();
3672 let err = executor
3673 .run(&workflow)
3674 .expect_err("unknown teacher candidate should fail before execution");
3675
3676 assert_eq!(
3677 err,
3678 WorkflowExecutionError::UnknownNodeReference {
3679 node: "teacher-review".to_string(),
3680 field: "candidates",
3681 reference: "candidate-b".to_string(),
3682 }
3683 );
3684 }
3685
3686 #[test]
3687 fn teacher_candidate_serialization() {
3688 let candidate = TeacherCandidate {
3689 candidate_id: "teacher-review:branch-a".to_string(),
3690 kind: TeacherCandidateKind::WorkflowRecipe,
3691 status: TeacherCandidateStatus::Proposed,
3692 source_node_id: "branch-a".to_string(),
3693 source_branch_id: Some("branch-a".to_string()),
3694 summary: "Winning branch found a reusable workflow recipe.".to_string(),
3695 evidence: vec![
3696 "status=Succeeded".to_string(),
3697 "tokens=42, cost_microusd=7".to_string(),
3698 ],
3699 replay_results: vec![StudentReplayResult {
3700 trace_id: "trace-a".to_string(),
3701 candidate_id: "teacher-review:branch-a".to_string(),
3702 baseline: StudentReplayMetrics {
3703 score: 70,
3704 cost_microusd: 10,
3705 },
3706 candidate: StudentReplayMetrics {
3707 score: 74,
3708 cost_microusd: 12,
3709 },
3710 required_tests: vec![StudentReplayTestResult {
3711 name: "cargo test -p codewhale-workflow".to_string(),
3712 passed: true,
3713 }],
3714 policy_violations: Vec::new(),
3715 stale: false,
3716 notes: Some("offline replay improved the constrained student".to_string()),
3717 }],
3718 };
3719
3720 let json = serde_json::to_string(&candidate).expect("serialize teacher candidate");
3721
3722 assert!(json.contains("\"kind\":\"workflow_recipe\""));
3723 assert!(json.contains("\"status\":\"proposed\""));
3724 assert!(json.contains("\"replay_results\""));
3725 let parsed: TeacherCandidate =
3726 serde_json::from_str(&json).expect("parse teacher candidate");
3727 assert_eq!(parsed, candidate);
3728 }
3729
3730 #[test]
3731 fn teacher_review_produces_candidate_from_trace() {
3732 let review = TeacherReviewSpec {
3733 id: "teacher-review".to_string(),
3734 candidates: vec!["winning-branch".to_string()],
3735 promotion_policy: PromotionPolicy::default(),
3736 };
3737 let execution = WorkflowExecution {
3738 branch_results: vec![BranchResult {
3739 branch_id: "winning-branch".to_string(),
3740 task_id: "winning-branch".to_string(),
3741 status: WorkflowRunStatus::Succeeded,
3742 usage: WorkflowUsage {
3743 input_tokens: 30,
3744 output_tokens: 12,
3745 cost_microusd: 7,
3746 },
3747 memo_usage: WorkflowMemoUsage::default(),
3748 artifacts: vec!["trace://branches/winning-branch".to_string()],
3749 notes: Some("branch produced a minimal verified patch".to_string()),
3750 }],
3751 ..WorkflowExecution::default()
3752 };
3753
3754 let report = TeacherReviewReport::from_execution(&review, &execution);
3755
3756 assert_eq!(report.review_node_id, "teacher-review");
3757 assert_eq!(report.candidates.len(), 1);
3758 assert_eq!(
3759 report.candidates[0].kind,
3760 TeacherCandidateKind::WorkflowRecipe
3761 );
3762 assert_eq!(
3763 report.candidates[0].status,
3764 TeacherCandidateStatus::Proposed
3765 );
3766 assert!(
3767 report.candidates[0]
3768 .evidence
3769 .iter()
3770 .any(|line| line.contains("tokens=42"))
3771 );
3772 }
3773
3774 #[test]
3775 fn failed_leaf_becomes_regression_test_candidate() {
3776 let review = TeacherReviewSpec {
3777 id: "teacher-review".to_string(),
3778 candidates: vec!["verify-failure".to_string()],
3779 promotion_policy: PromotionPolicy::default(),
3780 };
3781 let execution = WorkflowExecution {
3782 leaf_results: vec![LeafResult {
3783 leaf_id: "verify-failure".to_string(),
3784 task_id: "verify-failure".to_string(),
3785 role: None,
3786 profile: None,
3787 status: WorkflowRunStatus::Failed,
3788 usage: WorkflowUsage::default(),
3789 memo_usage: WorkflowMemoUsage::default(),
3790 output: Some("cargo test failed with a replay mismatch".to_string()),
3791 artifacts: Vec::new(),
3792 schema_error: None,
3793 }],
3794 ..WorkflowExecution::default()
3795 };
3796
3797 let candidates = teacher_candidates_from_execution(&review, &execution);
3798
3799 assert_eq!(candidates.len(), 1);
3800 assert_eq!(candidates[0].kind, TeacherCandidateKind::RegressionTest);
3801 assert_eq!(candidates[0].status, TeacherCandidateStatus::Proposed);
3802 assert!(
3803 candidates[0]
3804 .evidence
3805 .iter()
3806 .any(|line| { line.contains("cargo test failed with a replay mismatch") })
3807 );
3808 }
3809
3810 #[test]
3811 fn student_replay_promotes_only_on_delta() {
3812 let gate = PromotionGate {
3813 min_score_delta: 3,
3814 max_cost_delta_microusd: Some(25),
3815 ..PromotionGate::default()
3816 };
3817 let replay = StudentReplayResult {
3818 trace_id: "trace-a".to_string(),
3819 candidate_id: "teacher-review:branch-a".to_string(),
3820 baseline: StudentReplayMetrics {
3821 score: 80,
3822 cost_microusd: 100,
3823 },
3824 candidate: StudentReplayMetrics {
3825 score: 84,
3826 cost_microusd: 120,
3827 },
3828 required_tests: vec![StudentReplayTestResult {
3829 name: "workflow replay".to_string(),
3830 passed: true,
3831 }],
3832 policy_violations: Vec::new(),
3833 stale: false,
3834 notes: None,
3835 };
3836
3837 let promoted = gate.evaluate_replay("teacher-review:branch-a", &replay);
3838 assert!(promoted.promoted());
3839 assert_eq!(promoted.status, TeacherCandidateStatus::Promoted);
3840 assert_eq!(promoted.score_delta, 4);
3841
3842 let weak_replay = StudentReplayResult {
3843 candidate: StudentReplayMetrics {
3844 score: 82,
3845 cost_microusd: 120,
3846 },
3847 ..replay
3848 };
3849 let rejected = gate.evaluate_replay("teacher-review:branch-a", &weak_replay);
3850 assert!(!rejected.promoted());
3851 assert_eq!(rejected.status, TeacherCandidateStatus::Rejected);
3852 assert!(
3853 rejected
3854 .reasons
3855 .iter()
3856 .any(|reason| reason.contains("below required 3"))
3857 );
3858 }
3859
3860 #[test]
3861 fn promotion_gate_rejects_stale_policy_cost_and_failed_tests() {
3862 let gate = PromotionGate {
3863 min_score_delta: 1,
3864 max_cost_delta_microusd: Some(10),
3865 ..PromotionGate::default()
3866 };
3867 let replay = StudentReplayResult {
3868 trace_id: "trace-a".to_string(),
3869 candidate_id: "teacher-review:branch-a".to_string(),
3870 baseline: StudentReplayMetrics {
3871 score: 70,
3872 cost_microusd: 10,
3873 },
3874 candidate: StudentReplayMetrics {
3875 score: 90,
3876 cost_microusd: 30,
3877 },
3878 required_tests: vec![StudentReplayTestResult {
3879 name: "required regression".to_string(),
3880 passed: false,
3881 }],
3882 policy_violations: vec!["writes outside file scope".to_string()],
3883 stale: true,
3884 notes: None,
3885 };
3886
3887 let decision = gate.evaluate_replay("teacher-review:branch-a", &replay);
3888
3889 assert_eq!(decision.status, TeacherCandidateStatus::Rejected);
3890 assert!(
3891 decision
3892 .reasons
3893 .iter()
3894 .any(|reason| { reason.contains("cost delta 20 exceeds allowed 10") })
3895 );
3896 assert!(
3897 decision
3898 .reasons
3899 .iter()
3900 .any(|reason| { reason.contains("required test `required regression` failed") })
3901 );
3902 assert!(
3903 decision
3904 .reasons
3905 .iter()
3906 .any(|reason| { reason.contains("policy violation: writes outside file scope") })
3907 );
3908 assert!(
3909 decision
3910 .reasons
3911 .iter()
3912 .any(|reason| { reason.contains("student replay result is stale") })
3913 );
3914 }
3915
3916 #[test]
3917 fn promotion_gate_requires_recorded_replay_before_candidate_promotion() {
3918 let candidate = TeacherCandidate {
3919 candidate_id: "teacher-review:branch-a".to_string(),
3920 kind: TeacherCandidateKind::WorkflowRecipe,
3921 status: TeacherCandidateStatus::Proposed,
3922 source_node_id: "branch-a".to_string(),
3923 source_branch_id: Some("branch-a".to_string()),
3924 summary: "candidate waits for replay".to_string(),
3925 evidence: Vec::new(),
3926 replay_results: Vec::new(),
3927 };
3928
3929 let decision = PromotionGate::default().evaluate_candidate(&candidate);
3930
3931 assert_eq!(decision.status, TeacherCandidateStatus::Rejected);
3932 assert_eq!(
3933 decision.reasons,
3934 vec!["no student replay result recorded".to_string()]
3935 );
3936 }
3937
3938 #[test]
3939 fn tournament_selects_passing_minimal_branch() {
3940 let tournament = BranchTournament { min_score: 60 };
3941 let candidates = vec![
3942 candidate(
3943 "expensive-pass",
3944 WorkflowRunStatus::Succeeded,
3945 90,
3946 90,
3947 "quality",
3948 ),
3949 candidate("failed-cheap", WorkflowRunStatus::Failed, 100, 1, "broken"),
3950 candidate(
3951 "cheap-pass",
3952 WorkflowRunStatus::Succeeded,
3953 70,
3954 10,
3955 "minimal",
3956 ),
3957 candidate("too-low", WorkflowRunStatus::Succeeded, 40, 2, "weak"),
3958 ];
3959
3960 let selected = tournament
3961 .select(&candidates)
3962 .expect("one passing branch should be selected");
3963
3964 assert_eq!(selected.branch_id, "cheap-pass");
3965 }
3966
3967 #[test]
3968 fn pareto_frontier_keeps_diverse_candidates() {
3969 let frontier = ParetoFrontier { max_items: 4 };
3970 let candidates = vec![
3971 candidate("quality", WorkflowRunStatus::Succeeded, 95, 100, "quality"),
3972 candidate("minimal", WorkflowRunStatus::Succeeded, 70, 10, "small"),
3973 candidate("dominated", WorkflowRunStatus::Succeeded, 60, 40, "middle"),
3974 candidate("failed", WorkflowRunStatus::Failed, 100, 1, "broken"),
3975 ];
3976
3977 let selected = frontier.select(&candidates);
3978
3979 assert_eq!(
3980 selected
3981 .iter()
3982 .map(|candidate| candidate.branch_id.as_str())
3983 .collect::<Vec<_>>(),
3984 vec!["quality", "minimal"]
3985 );
3986 assert_eq!(
3987 selected
3988 .iter()
3989 .filter_map(|candidate| candidate.diversity_key.as_deref())
3990 .collect::<Vec<_>>(),
3991 vec!["quality", "small"]
3992 );
3993 }
3994}