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_scope(left);
2061 let right = normalize_scope(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
2076fn normalize_scope(scope: &str) -> String {
2077 let trimmed = scope.trim().trim_start_matches("./").trim_end_matches('/');
2078 trimmed
2079 .strip_suffix("/**")
2080 .or_else(|| trimmed.strip_suffix("/*"))
2081 .unwrap_or(trimmed)
2082 .to_string()
2083}
2084
2085fn glob_prefix(scope: &str) -> String {
2086 scope
2087 .split('*')
2088 .next()
2089 .unwrap_or(scope)
2090 .trim_end_matches('/')
2091 .to_string()
2092}
2093
2094#[cfg(test)]
2095mod tests {
2096 use super::*;
2097
2098 fn task(id: &str) -> Task {
2099 Task {
2100 id: id.to_string(),
2101 prompt: format!("run {id}"),
2102 agent_type: AgentType::General,
2103 mode: TaskMode::ReadOnly,
2104 isolation: IsolationMode::Shared,
2105 file_scope: Vec::new(),
2106 depends_on_results: Vec::new(),
2107 max_steps: None,
2108 timeout_secs: None,
2109 }
2110 }
2111
2112 fn config(phases: Vec<Phase>) -> WorkflowConfig {
2113 WorkflowConfig {
2114 goal: "cache-change".to_string(),
2115 max_concurrent: 4,
2116 description: None,
2117 phases,
2118 }
2119 }
2120
2121 fn phase(name: &str, depends_on: &[&str], tasks: Vec<Task>) -> Phase {
2122 Phase {
2123 name: name.to_string(),
2124 description: None,
2125 depends_on: depends_on.iter().map(|value| value.to_string()).collect(),
2126 parallel: false,
2127 on_failure: FailurePolicy::SkipContinue,
2128 tasks,
2129 }
2130 }
2131
2132 fn leaf_node(id: &str) -> WorkflowNode {
2133 WorkflowNode::Leaf(LeafSpec {
2134 id: id.to_string(),
2135 prompt: format!("run {id}"),
2136 agent_type: AgentType::General,
2137 role: None,
2138 profile: None,
2139 mode: TaskMode::ReadOnly,
2140 isolation: IsolationMode::Shared,
2141 file_scope: Vec::new(),
2142 depends_on_results: Vec::new(),
2143 budget: BudgetSpec::default(),
2144 permissions: PermissionSpec::default(),
2145 model_policy: ModelPolicy::default(),
2146 })
2147 }
2148
2149 fn leaf_node_with_budget(id: &str, budget: BudgetSpec) -> WorkflowNode {
2150 WorkflowNode::Leaf(LeafSpec {
2151 id: id.to_string(),
2152 prompt: format!("run {id}"),
2153 agent_type: AgentType::General,
2154 role: None,
2155 profile: None,
2156 mode: TaskMode::ReadOnly,
2157 isolation: IsolationMode::Shared,
2158 file_scope: Vec::new(),
2159 depends_on_results: Vec::new(),
2160 budget,
2161 permissions: PermissionSpec::default(),
2162 model_policy: ModelPolicy::default(),
2163 })
2164 }
2165
2166 fn invalid_leaf_node(id: &str) -> WorkflowNode {
2167 WorkflowNode::Leaf(LeafSpec {
2168 id: id.to_string(),
2169 prompt: " ".to_string(),
2170 agent_type: AgentType::General,
2171 role: None,
2172 profile: None,
2173 mode: TaskMode::ReadOnly,
2174 isolation: IsolationMode::Shared,
2175 file_scope: Vec::new(),
2176 depends_on_results: Vec::new(),
2177 budget: BudgetSpec::default(),
2178 permissions: PermissionSpec::default(),
2179 model_policy: ModelPolicy::default(),
2180 })
2181 }
2182
2183 fn workflow_spec(nodes: Vec<WorkflowNode>) -> WorkflowSpec {
2184 WorkflowSpec {
2185 id: Some("mock-workflow".to_string()),
2186 goal: "prove mock executor control flow".to_string(),
2187 description: None,
2188 budget: BudgetSpec::default(),
2189 permissions: PermissionSpec::default(),
2190 model_policy: ModelPolicy::default(),
2191 promotion_policy: PromotionPolicy::default(),
2192 gates: Vec::new(),
2193 nodes,
2194 }
2195 }
2196
2197 fn control_result<'a>(
2198 execution: &'a WorkflowExecution,
2199 node_id: &str,
2200 ) -> &'a ControlNodeResult {
2201 execution
2202 .control_node_results
2203 .iter()
2204 .find(|result| result.node_id == node_id)
2205 .expect("control node result should exist")
2206 }
2207
2208 fn candidate(
2209 branch_id: &str,
2210 status: WorkflowRunStatus,
2211 score: u32,
2212 cost: u64,
2213 diversity_key: &str,
2214 ) -> BranchCandidate {
2215 BranchCandidate {
2216 branch_id: branch_id.to_string(),
2217 status,
2218 score,
2219 cost,
2220 diversity_key: Some(diversity_key.to_string()),
2221 }
2222 }
2223
2224 #[test]
2225 fn independent_phases_preserve_declaration_order() {
2226 let workflow = config(vec![
2227 phase("discover", &[], vec![task("scan")]),
2228 phase("report", &[], vec![task("summarize")]),
2229 ]);
2230
2231 let plan = workflow.compile().expect("workflow should compile");
2232
2233 assert_eq!(
2234 plan.phase_names().collect::<Vec<_>>(),
2235 vec!["discover", "report"]
2236 );
2237 }
2238
2239 #[test]
2240 fn dependencies_override_declaration_order_deterministically() {
2241 let workflow = config(vec![
2242 phase("review", &["implement"], vec![task("review-results")]),
2243 phase("discover", &[], vec![task("scan")]),
2244 phase("implement", &["discover"], vec![task("patch")]),
2245 phase("report", &["review"], vec![task("summarize")]),
2246 ]);
2247
2248 let plan = workflow.compile().expect("workflow should compile");
2249
2250 assert_eq!(
2251 plan.phase_names().collect::<Vec<_>>(),
2252 vec!["discover", "implement", "review", "report"]
2253 );
2254 }
2255
2256 #[test]
2257 fn rejects_empty_workflow() {
2258 let err = config(Vec::new())
2259 .validate()
2260 .expect_err("empty workflow should fail");
2261
2262 assert_eq!(err, WorkflowValidationError::EmptyWorkflow);
2263 }
2264
2265 #[test]
2266 fn rejects_empty_phase() {
2267 let err = config(vec![phase("empty", &[], Vec::new())])
2268 .validate()
2269 .expect_err("empty phase should fail");
2270
2271 assert_eq!(
2272 err,
2273 WorkflowValidationError::EmptyPhase {
2274 phase: "empty".to_string()
2275 }
2276 );
2277 }
2278
2279 #[test]
2280 fn rejects_invalid_max_concurrent() {
2281 let mut workflow = config(vec![phase("discover", &[], vec![task("scan")])]);
2282 workflow.max_concurrent = 0;
2283
2284 let err = workflow
2285 .validate()
2286 .expect_err("zero concurrency should fail");
2287
2288 assert_eq!(
2289 err,
2290 WorkflowValidationError::InvalidMaxConcurrent { value: 0 }
2291 );
2292 }
2293
2294 #[test]
2295 fn rejects_duplicate_phase_names() {
2296 let err = config(vec![
2297 phase("discover", &[], vec![task("scan")]),
2298 phase("discover", &[], vec![task("scan-again")]),
2299 ])
2300 .validate()
2301 .expect_err("duplicate phase should fail");
2302
2303 assert!(matches!(
2304 err,
2305 WorkflowValidationError::DuplicatePhase { .. }
2306 ));
2307 }
2308
2309 #[test]
2310 fn rejects_duplicate_task_ids() {
2311 let err = config(vec![
2312 phase("discover", &[], vec![task("scan")]),
2313 phase("report", &[], vec![task("scan")]),
2314 ])
2315 .validate()
2316 .expect_err("duplicate task should fail");
2317
2318 assert!(matches!(err, WorkflowValidationError::DuplicateTask { .. }));
2319 }
2320
2321 #[test]
2322 fn rejects_unknown_phase_dependency() {
2323 let err = config(vec![phase("report", &["missing"], vec![task("summarize")])])
2324 .validate()
2325 .expect_err("unknown dependency should fail");
2326
2327 assert!(matches!(
2328 err,
2329 WorkflowValidationError::InvalidPhaseDependency { .. }
2330 ));
2331 }
2332
2333 #[test]
2334 fn rejects_phase_dependency_cycles() {
2335 let workflow = config(vec![
2336 phase("a", &["b"], vec![task("a-task")]),
2337 phase("b", &["a"], vec![task("b-task")]),
2338 ]);
2339
2340 let err = workflow.validate().expect_err("cycle should fail");
2341
2342 assert!(matches!(
2343 err,
2344 WorkflowValidationError::PhaseDependencyCycle { .. }
2345 ));
2346 }
2347
2348 #[test]
2349 fn rejects_task_result_dependency_from_same_parallel_phase() {
2350 let mut first = task("first");
2351 first.depends_on_results.push("second".to_string());
2352 let mut parallel = phase("parallel", &[], vec![first, task("second")]);
2353 parallel.parallel = true;
2354
2355 let err = config(vec![parallel])
2356 .validate()
2357 .expect_err("same-phase result dependency should fail");
2358
2359 assert!(matches!(
2360 err,
2361 WorkflowValidationError::UnavailableTaskResultDependency { .. }
2362 ));
2363 }
2364
2365 #[test]
2366 fn rejects_task_result_dependency_from_later_phase() {
2367 let mut summarize = task("summarize");
2368 summarize.depends_on_results.push("scan".to_string());
2369 let workflow = config(vec![
2370 phase("report", &[], vec![summarize]),
2371 phase("discover", &[], vec![task("scan")]),
2372 ]);
2373
2374 let err = workflow
2375 .validate()
2376 .expect_err("later-phase result dependency should fail");
2377
2378 assert!(matches!(
2379 err,
2380 WorkflowValidationError::UnavailableTaskResultDependency { .. }
2381 ));
2382 }
2383
2384 #[test]
2385 fn allows_task_result_dependency_from_earlier_phase() {
2386 let upstream = phase("discover", &[], vec![task("scan")]);
2387 let mut summarize = task("summarize");
2388 summarize.depends_on_results.push("scan".to_string());
2389 let downstream = phase("report", &["discover"], vec![summarize]);
2390
2391 config(vec![upstream, downstream])
2392 .validate()
2393 .expect("earlier-phase result should be available");
2394 }
2395
2396 #[test]
2397 fn rejects_parallel_read_write_without_file_scope() {
2398 let mut write = task("write");
2399 write.mode = TaskMode::ReadWrite;
2400 let mut parallel = phase("parallel", &[], vec![write]);
2401 parallel.parallel = true;
2402
2403 let err = config(vec![parallel])
2404 .validate()
2405 .expect_err("write task needs a scope");
2406
2407 assert!(matches!(
2408 err,
2409 WorkflowValidationError::MissingParallelWriteScope { .. }
2410 ));
2411 }
2412
2413 #[test]
2414 fn detects_overlapping_parallel_write_scopes_with_path_boundaries() {
2415 let mut left = task("auth");
2416 left.mode = TaskMode::ReadWrite;
2417 left.file_scope = vec!["src/auth/**".to_string()];
2418 let mut right = task("auth-login");
2419 right.mode = TaskMode::ReadWrite;
2420 right.file_scope = vec!["src/auth/login.rs".to_string()];
2421 let mut parallel = phase("parallel", &[], vec![left, right]);
2422 parallel.parallel = true;
2423
2424 let err = config(vec![parallel])
2425 .validate()
2426 .expect_err("nested scopes should overlap");
2427
2428 assert!(matches!(
2429 err,
2430 WorkflowValidationError::OverlappingParallelWriteScope { .. }
2431 ));
2432 }
2433
2434 #[test]
2435 fn does_not_confuse_path_prefixes_for_overlapping_scopes() {
2436 let mut left = task("auth");
2437 left.mode = TaskMode::ReadWrite;
2438 left.file_scope = vec!["src/auth/**".to_string()];
2439 let mut right = task("auth-admin");
2440 right.mode = TaskMode::ReadWrite;
2441 right.file_scope = vec!["src/auth_admin/**".to_string()];
2442 let mut parallel = phase("parallel", &[], vec![left, right]);
2443 parallel.parallel = true;
2444
2445 config(vec![parallel])
2446 .validate()
2447 .expect("component boundary scopes should not overlap");
2448 }
2449
2450 #[test]
2451 fn json_roundtrip_keeps_snake_case_enum_names() {
2452 let mut task = task("patch");
2453 task.agent_type = AgentType::Implementer;
2454 task.mode = TaskMode::ReadWrite;
2455 task.isolation = IsolationMode::Worktree;
2456 task.file_scope = vec!["src/auth/**".to_string()];
2457 let mut parallel = phase("implement", &[], vec![task]);
2458 parallel.parallel = true;
2459 parallel.on_failure = FailurePolicy::Abort;
2460 let workflow = config(vec![parallel]);
2461
2462 let json = serde_json::to_string(&workflow).expect("serialize workflow");
2463
2464 assert!(json.contains("\"agent_type\":\"implementer\""));
2465 assert!(json.contains("\"mode\":\"read_write\""));
2466 assert!(json.contains("\"isolation\":\"worktree\""));
2467 assert!(json.contains("\"on_failure\":\"abort\""));
2468
2469 let parsed: WorkflowConfig = serde_json::from_str(&json).expect("parse workflow");
2470 assert_eq!(parsed, workflow);
2471 }
2472
2473 #[test]
2474 fn isolation_auto_defaults_parallel_write_to_worktree() {
2475 assert_eq!(IsolationMode::default(), IsolationMode::Auto);
2476 assert_eq!(
2477 IsolationMode::Auto.resolve(true),
2478 IsolationMode::Worktree
2479 );
2480 assert_eq!(
2481 IsolationMode::Auto.resolve(false),
2482 IsolationMode::Shared
2483 );
2484 assert_eq!(
2486 IsolationMode::Shared.resolve(true),
2487 IsolationMode::Shared
2488 );
2489 assert!(IsolationMode::Worktree.wants_worktree(false));
2490 assert!(!IsolationMode::Shared.wants_worktree(true));
2491 }
2492
2493 #[test]
2494 fn leaf_write_capable_and_worktree_defaults() {
2495 let read_only = LeafSpec {
2496 id: "ro".to_string(),
2497 prompt: "inspect".to_string(),
2498 agent_type: AgentType::Explore,
2499 role: None,
2500 profile: None,
2501 mode: TaskMode::ReadOnly,
2502 isolation: IsolationMode::Auto,
2503 file_scope: Vec::new(),
2504 depends_on_results: Vec::new(),
2505 budget: BudgetSpec::default(),
2506 permissions: PermissionSpec::default(),
2507 model_policy: ModelPolicy::default(),
2508 };
2509 assert!(!leaf_is_write_capable(&read_only));
2510 assert!(!leaf_wants_worktree(&read_only, true));
2511
2512 let mut read_only_implementer = read_only.clone();
2513 read_only_implementer.id = "ro-implementer".to_string();
2514 read_only_implementer.agent_type = AgentType::Implementer;
2515 assert!(
2516 !leaf_is_write_capable(&read_only_implementer),
2517 "role identity must not grant write authority"
2518 );
2519 assert!(
2520 !leaf_wants_worktree(&read_only_implementer, true),
2521 "parallel read-only implementers stay shared under auto isolation"
2522 );
2523
2524 let mut write = read_only.clone();
2525 write.id = "rw".to_string();
2526 write.mode = TaskMode::ReadWrite;
2527 write.agent_type = AgentType::Implementer;
2528 assert!(leaf_is_write_capable(&write));
2529 assert!(leaf_wants_worktree(&write, true));
2531 assert!(!leaf_wants_worktree(&write, false));
2533
2534 write.isolation = IsolationMode::Shared;
2535 assert!(
2536 !leaf_wants_worktree(&write, true),
2537 "explicit shared is the same-worktree override"
2538 );
2539
2540 write.isolation = IsolationMode::Worktree;
2541 assert!(leaf_wants_worktree(&write, true));
2542 assert!(leaf_wants_worktree(&write, false));
2543 }
2544
2545 #[test]
2546 fn workflow_ir_roundtrip() {
2547 let discover_leaf = LeafSpec {
2548 id: "scan-readme".to_string(),
2549 prompt: "Inspect README setup gaps".to_string(),
2550 agent_type: AgentType::Explore,
2551 role: None,
2552 profile: Some("scout".to_string()),
2553 mode: TaskMode::ReadOnly,
2554 isolation: IsolationMode::Shared,
2555 file_scope: vec!["README.md".to_string()],
2556 depends_on_results: Vec::new(),
2557 budget: BudgetSpec {
2558 max_steps: Some(8),
2559 timeout_secs: Some(300),
2560 max_parallel: None,
2561 max_tokens: None,
2562 },
2563 permissions: PermissionSpec::default(),
2564 model_policy: ModelPolicy {
2565 provider: Some("openai".to_string()),
2566 model: Some("gpt-5.4".to_string()),
2567 fallback_models: Vec::new(),
2568 },
2569 };
2570 let workflow = WorkflowSpec {
2571 id: Some("v090-readme-check".to_string()),
2572 goal: "tighten setup docs".to_string(),
2573 description: Some("metadata-only typed Workflow IR".to_string()),
2574 budget: BudgetSpec {
2575 max_steps: Some(30),
2576 timeout_secs: Some(1_800),
2577 max_parallel: Some(2),
2578 max_tokens: None,
2579 },
2580 permissions: PermissionSpec {
2581 allow_write: false,
2582 allow_network: false,
2583 deny_all_tools: false,
2584 allowed_tools: vec!["rg".to_string()],
2585 file_scope: vec!["README.md".to_string()],
2586 },
2587 model_policy: ModelPolicy {
2588 provider: Some("openai".to_string()),
2589 model: Some("gpt-5.4".to_string()),
2590 fallback_models: vec!["gpt-5.4-mini".to_string()],
2591 },
2592 promotion_policy: PromotionPolicy {
2593 strategy: PromotionStrategy::TeacherSelected,
2594 require_teacher_review: true,
2595 min_successful_branches: Some(1),
2596 promotion_gate: PromotionGate::default(),
2597 },
2598 gates: Vec::new(),
2599 nodes: vec![
2600 WorkflowNode::BranchSet(BranchSpec {
2601 id: "discover".to_string(),
2602 description: Some("parallel doc inspection".to_string()),
2603 parallel: true,
2604 budget: BudgetSpec {
2605 max_steps: Some(12),
2606 timeout_secs: Some(600),
2607 max_parallel: Some(2),
2608 max_tokens: None,
2609 },
2610 permissions: PermissionSpec::default(),
2611 model_policy: ModelPolicy::default(),
2612 children: vec![WorkflowNode::Leaf(discover_leaf)],
2613 }),
2614 WorkflowNode::Sequence(SequenceSpec {
2615 id: "review-and-reduce".to_string(),
2616 children: vec![
2617 WorkflowNode::TeacherReview(TeacherReviewSpec {
2618 id: "select-best".to_string(),
2619 candidates: vec!["scan-readme".to_string()],
2620 promotion_policy: PromotionPolicy {
2621 strategy: PromotionStrategy::BestScore,
2622 require_teacher_review: true,
2623 min_successful_branches: Some(1),
2624 promotion_gate: PromotionGate::default(),
2625 },
2626 }),
2627 WorkflowNode::Reduce(ReduceSpec {
2628 id: "summarize".to_string(),
2629 inputs: vec!["scan-readme".to_string()],
2630 prompt: "Summarize the smallest safe patch".to_string(),
2631 model_policy: ModelPolicy::default(),
2632 }),
2633 ],
2634 }),
2635 WorkflowNode::Cond(CondSpec {
2636 id: "maybe-expand".to_string(),
2637 condition: "summary identifies multiple independent gaps".to_string(),
2638 then_nodes: vec![WorkflowNode::Expand(ExpandSpec {
2639 id: "split-followups".to_string(),
2640 source: "summarize".to_string(),
2641 max_children: None,
2642 template: Some(Box::new(WorkflowNode::Leaf(LeafSpec {
2643 id: "followup-template".to_string(),
2644 prompt: "Patch one independent gap".to_string(),
2645 agent_type: AgentType::Implementer,
2646 role: None,
2647 profile: None,
2648 mode: TaskMode::ReadWrite,
2649 isolation: IsolationMode::Worktree,
2650 file_scope: vec!["README.md".to_string()],
2651 depends_on_results: Vec::new(),
2652 budget: BudgetSpec::default(),
2653 permissions: PermissionSpec {
2654 allow_write: true,
2655 allow_network: false,
2656 deny_all_tools: false,
2657 allowed_tools: Vec::new(),
2658 file_scope: vec!["README.md".to_string()],
2659 },
2660 model_policy: ModelPolicy::default(),
2661 }))),
2662 })],
2663 else_nodes: vec![WorkflowNode::LoopUntil(LoopUntilSpec {
2664 id: "verify-once".to_string(),
2665 condition: "local verification passes".to_string(),
2666 max_iterations: Some(1),
2667 children: Vec::new(),
2668 })],
2669 }),
2670 ],
2671 };
2672
2673 let json = serde_json::to_string_pretty(&workflow).expect("serialize workflow ir");
2674
2675 assert!(json.contains("\"kind\": \"branch_set\""));
2676 assert!(json.contains("\"strategy\": \"teacher_selected\""));
2677 assert!(json.contains("\"profile\": \"scout\""));
2678 let parsed: WorkflowSpec = serde_json::from_str(&json).expect("parse workflow ir");
2679 assert_eq!(parsed, workflow);
2680
2681 let minimal: WorkflowSpec = serde_json::from_str(r#"{"goal":"ship v0.9","nodes":[]}"#)
2682 .expect("parse minimal workflow ir");
2683 assert_eq!(minimal.budget, BudgetSpec::default());
2684 assert_eq!(minimal.permissions, PermissionSpec::default());
2685 assert_eq!(minimal.model_policy, ModelPolicy::default());
2686
2687 let legacy_leaf: LeafSpec = serde_json::from_str(r#"{"id":"scan","prompt":"scan safely"}"#)
2689 .expect("parse pre-profile leaf ir");
2690 assert_eq!(legacy_leaf.profile, None);
2691 let legacy_json = serde_json::to_string(&legacy_leaf).expect("serialize legacy leaf");
2692 assert!(!legacy_json.contains("profile"));
2693 }
2694
2695 #[test]
2696 fn fleet_validation_accepts_one_hundred_agents_and_variable_models() {
2697 let nodes = (0..DEFAULT_FLEET_WORKFLOW_MAX_AGENTS)
2698 .map(|index| {
2699 let mut leaf = match leaf_node(&format!("agent-{index}")) {
2700 WorkflowNode::Leaf(leaf) => leaf,
2701 _ => unreachable!("leaf helper returns a leaf"),
2702 };
2703 leaf.model_policy = if index == 0 {
2704 ModelPolicy {
2705 provider: Some("deepseek".to_string()),
2706 model: Some("deepseek-v4-pro".to_string()),
2707 fallback_models: Vec::new(),
2708 }
2709 } else {
2710 ModelPolicy {
2711 provider: Some("deepseek".to_string()),
2712 model: Some("deepseek-v4-flash".to_string()),
2713 fallback_models: Vec::new(),
2714 }
2715 };
2716 WorkflowNode::Leaf(leaf)
2717 })
2718 .collect();
2719 let workflow = workflow_spec(nodes);
2720
2721 let shape = workflow
2722 .validate_for_fleet()
2723 .expect("one hundred agents should fit the Fleet Workflow limit");
2724
2725 assert_eq!(shape.total_agents, DEFAULT_FLEET_WORKFLOW_MAX_AGENTS);
2726 assert_eq!(shape.max_depth, 1);
2727 }
2728
2729 #[test]
2730 fn fleet_validation_rejects_more_than_one_hundred_agents() {
2731 let nodes = (0..=DEFAULT_FLEET_WORKFLOW_MAX_AGENTS)
2732 .map(|index| leaf_node(&format!("agent-{index}")))
2733 .collect();
2734 let workflow = workflow_spec(nodes);
2735
2736 let err = workflow
2737 .validate_for_fleet()
2738 .expect_err("agent population should be bounded before Fleet launch");
2739
2740 assert_eq!(
2741 err,
2742 WorkflowFleetLimitError::TooManyAgents {
2743 total_agents: DEFAULT_FLEET_WORKFLOW_MAX_AGENTS + 1,
2744 max_total_agents: DEFAULT_FLEET_WORKFLOW_MAX_AGENTS,
2745 }
2746 );
2747 }
2748
2749 #[test]
2750 fn fleet_validation_rejects_depth_beyond_five() {
2751 let mut node = leaf_node("deep-leaf");
2752 for depth in (0..DEFAULT_FLEET_WORKFLOW_MAX_DEPTH).rev() {
2753 node = WorkflowNode::BranchSet(BranchSpec {
2754 id: format!("ring-{depth}"),
2755 description: None,
2756 parallel: true,
2757 budget: BudgetSpec::default(),
2758 permissions: PermissionSpec::default(),
2759 model_policy: ModelPolicy::default(),
2760 children: vec![node],
2761 });
2762 }
2763 let workflow = workflow_spec(vec![node]);
2764
2765 let err = workflow
2766 .validate_for_fleet()
2767 .expect_err("sixth agent ring should be rejected");
2768
2769 assert_eq!(
2770 err,
2771 WorkflowFleetLimitError::RecursionTooDeep {
2772 depth: DEFAULT_FLEET_WORKFLOW_MAX_DEPTH + 1,
2773 max_depth: DEFAULT_FLEET_WORKFLOW_MAX_DEPTH,
2774 }
2775 );
2776 }
2777
2778 #[test]
2779 fn fleet_validation_counts_loop_and_expand_fanout_conservatively() {
2780 let workflow = workflow_spec(vec![
2781 WorkflowNode::LoopUntil(LoopUntilSpec {
2782 id: "retry-ring".to_string(),
2783 condition: "verifier passes".to_string(),
2784 max_iterations: Some(3),
2785 children: vec![leaf_node("retry-worker")],
2786 }),
2787 WorkflowNode::Expand(ExpandSpec {
2788 id: "split".to_string(),
2789 source: "retry-ring".to_string(),
2790 max_children: Some(4),
2791 template: Some(Box::new(leaf_node("split-template"))),
2792 }),
2793 ]);
2794
2795 let shape = workflow
2796 .validate_for_fleet()
2797 .expect("bounded loop and expand should validate");
2798
2799 assert_eq!(shape.total_agents, 7);
2800 assert_eq!(shape.max_depth, 1);
2801 }
2802
2803 #[test]
2804 fn fleet_validation_rejects_unbounded_loop_or_expand_before_launch() {
2805 let workflow = workflow_spec(vec![
2806 WorkflowNode::LoopUntil(LoopUntilSpec {
2807 id: "retry-ring".to_string(),
2808 condition: "verifier passes".to_string(),
2809 max_iterations: None,
2810 children: vec![leaf_node("retry-worker")],
2811 }),
2812 WorkflowNode::Expand(ExpandSpec {
2813 id: "split".to_string(),
2814 source: "retry-ring".to_string(),
2815 max_children: Some(4),
2816 template: Some(Box::new(leaf_node("split-template"))),
2817 }),
2818 ]);
2819
2820 assert!(matches!(
2821 workflow.validate_for_fleet(),
2822 Err(WorkflowFleetLimitError::UnboundedLoop { node }) if node == "retry-ring"
2823 ));
2824
2825 let workflow = workflow_spec(vec![WorkflowNode::Expand(ExpandSpec {
2826 id: "split".to_string(),
2827 source: "retry-ring".to_string(),
2828 max_children: None,
2829 template: Some(Box::new(leaf_node("split-template"))),
2830 })]);
2831
2832 assert!(matches!(
2833 workflow.validate_for_fleet(),
2834 Err(WorkflowFleetLimitError::UnboundedExpand { node }) if node == "split"
2835 ));
2836 }
2837
2838 #[test]
2839 fn branch_result_serialization() {
2840 let result = BranchResult {
2841 branch_id: "discover".to_string(),
2842 task_id: "scan".to_string(),
2843 status: WorkflowRunStatus::Succeeded,
2844 usage: WorkflowUsage {
2845 input_tokens: 100,
2846 output_tokens: 25,
2847 cost_microusd: 42,
2848 },
2849 memo_usage: WorkflowMemoUsage::default(),
2850 artifacts: vec!["trace://branches/discover".to_string()],
2851 notes: Some("validated prompt surfaces".to_string()),
2852 };
2853
2854 let json = serde_json::to_string(&result).expect("serialize branch result");
2855
2856 assert!(json.contains("\"status\":\"succeeded\""));
2857 assert!(json.contains("\"cost_microusd\":42"));
2858 let parsed: BranchResult = serde_json::from_str(&json).expect("parse branch result");
2859 assert_eq!(parsed, result);
2860
2861 let minimal: BranchResult =
2862 serde_json::from_str(r#"{"branch_id":"discover","task_id":"scan","status":"pending"}"#)
2863 .expect("parse minimal branch result");
2864 assert_eq!(minimal.usage, WorkflowUsage::default());
2865 assert_eq!(minimal.memo_usage, WorkflowMemoUsage::default());
2866 assert!(minimal.artifacts.is_empty());
2867 assert_eq!(minimal.notes, None);
2868 }
2869
2870 #[test]
2871 fn leaf_result_serialization() {
2872 let result = LeafResult {
2873 leaf_id: "scan-readme".to_string(),
2874 task_id: "scan".to_string(),
2875 role: None,
2876 profile: Some("reviewer".to_string()),
2877 status: WorkflowRunStatus::Failed,
2878 usage: WorkflowUsage {
2879 input_tokens: 11,
2880 output_tokens: 7,
2881 cost_microusd: 3,
2882 },
2883 memo_usage: WorkflowMemoUsage {
2884 armh_hits: 1,
2885 armh_misses: 0,
2886 armh_saved_estimated_tokens: 128,
2887 provider_prompt_cache_hits: 2,
2888 provider_prompt_cache_misses: 1,
2889 },
2890 output: Some("README needs clearer setup steps".to_string()),
2891 artifacts: vec!["trace://leaves/scan-readme".to_string()],
2892 schema_error: None,
2893 };
2894
2895 let json = serde_json::to_string(&result).expect("serialize leaf result");
2896
2897 assert!(json.contains("\"status\":\"failed\""));
2898 assert!(json.contains("\"input_tokens\":11"));
2899 assert!(json.contains("\"armh_saved_estimated_tokens\":128"));
2900 assert!(json.contains("\"profile\":\"reviewer\""));
2901 let parsed: LeafResult = serde_json::from_str(&json).expect("parse leaf result");
2902 assert_eq!(parsed, result);
2903
2904 let minimal: LeafResult = serde_json::from_str(
2905 r#"{"leaf_id":"scan-readme","task_id":"scan","status":"pending"}"#,
2906 )
2907 .expect("parse minimal leaf result");
2908 assert_eq!(minimal.profile, None);
2909 assert_eq!(minimal.usage, WorkflowUsage::default());
2910 assert_eq!(minimal.memo_usage, WorkflowMemoUsage::default());
2911 assert_eq!(minimal.output, None);
2912 assert!(minimal.artifacts.is_empty());
2913 }
2914
2915 #[test]
2916 fn control_node_result_serialization() {
2917 let result = ControlNodeResult {
2918 node_id: "select-fix".to_string(),
2919 kind: ControlNodeKind::TeacherReview,
2920 status: WorkflowRunStatus::Running,
2921 selected_children: vec!["branch-a".to_string(), "branch-c".to_string()],
2922 summary: Some("teacher review is waiting on verifier evidence".to_string()),
2923 };
2924
2925 let json = serde_json::to_string(&result).expect("serialize control node result");
2926
2927 assert!(json.contains("\"kind\":\"teacher_review\""));
2928 assert!(json.contains("\"status\":\"running\""));
2929 let parsed: ControlNodeResult =
2930 serde_json::from_str(&json).expect("parse control node result");
2931 assert_eq!(parsed, result);
2932
2933 let minimal: ControlNodeResult = serde_json::from_str(
2934 r#"{"node_id":"select-fix","kind":"branch_set","status":"pending"}"#,
2935 )
2936 .expect("parse minimal control node result");
2937 assert!(minimal.selected_children.is_empty());
2938 assert_eq!(minimal.summary, None);
2939 }
2940
2941 #[test]
2942 fn run_mock_three_branch_workflow() {
2943 let workflow = workflow_spec(vec![WorkflowNode::BranchSet(BranchSpec {
2944 id: "discover".to_string(),
2945 description: None,
2946 parallel: true,
2947 budget: BudgetSpec::default(),
2948 permissions: PermissionSpec::default(),
2949 model_policy: ModelPolicy::default(),
2950 children: vec![
2951 leaf_node("scan-readme"),
2952 leaf_node("scan-config"),
2953 leaf_node("scan-tests"),
2954 ],
2955 })]);
2956
2957 let mut executor = MockWorkflowExecutor::new();
2958 let execution = executor.run(&workflow).expect("mock workflow should run");
2959
2960 assert_eq!(execution.status, WorkflowRunStatus::Succeeded);
2961 assert_eq!(
2962 execution
2963 .leaf_results
2964 .iter()
2965 .map(|result| result.leaf_id.as_str())
2966 .collect::<Vec<_>>(),
2967 vec!["scan-readme", "scan-config", "scan-tests"]
2968 );
2969 assert_eq!(execution.branch_results.len(), 1);
2970 assert_eq!(execution.branch_results[0].branch_id, "discover");
2971 assert_eq!(
2972 control_result(&execution, "discover").selected_children,
2973 vec!["scan-readme", "scan-config", "scan-tests"]
2974 );
2975 }
2976
2977 #[test]
2978 fn mock_executor_surfaces_leaf_profile() {
2979 let mut profiled_leaf = match leaf_node("review-change") {
2980 WorkflowNode::Leaf(leaf) => leaf,
2981 _ => unreachable!("leaf helper returns a leaf"),
2982 };
2983 profiled_leaf.profile = Some("reviewer".to_string());
2984 let workflow = workflow_spec(vec![
2985 WorkflowNode::Leaf(profiled_leaf),
2986 leaf_node("scan-readme"),
2987 ]);
2988
2989 let execution = MockWorkflowExecutor::new()
2990 .run(&workflow)
2991 .expect("mock workflow should run");
2992
2993 assert_eq!(execution.status, WorkflowRunStatus::Succeeded);
2994 assert_eq!(
2995 execution.leaf_results[0].profile.as_deref(),
2996 Some("reviewer")
2997 );
2998 assert_eq!(execution.leaf_results[1].profile, None);
2999 }
3000
3001 #[test]
3002 fn mock_executor_surfaces_leaf_role() {
3003 let mut role_leaf = match leaf_node("scout-issue") {
3004 WorkflowNode::Leaf(leaf) => leaf,
3005 _ => unreachable!("leaf helper returns a leaf"),
3006 };
3007 role_leaf.role = Some("scout".to_string());
3008 let workflow = workflow_spec(vec![WorkflowNode::Leaf(role_leaf)]);
3009
3010 let execution = MockWorkflowExecutor::new()
3011 .run(&workflow)
3012 .expect("mock workflow should run");
3013
3014 assert_eq!(execution.leaf_results[0].role.as_deref(), Some("scout"));
3015 }
3016
3017 #[test]
3018 fn leaf_role_token_rule_rejects_invalid_names() {
3019 for bad in ["", "has space", "role=scout"] {
3020 let mut leaf = match leaf_node("scan") {
3021 WorkflowNode::Leaf(leaf) => leaf,
3022 _ => unreachable!("leaf helper returns a leaf"),
3023 };
3024 leaf.role = Some(bad.to_string());
3025 let workflow = workflow_spec(vec![WorkflowNode::Leaf(leaf)]);
3026
3027 let err = MockWorkflowExecutor::new()
3028 .run(&workflow)
3029 .expect_err("invalid role token should fail validation");
3030
3031 assert!(
3032 matches!(&err, WorkflowExecutionError::InvalidLeafRole { role, .. } if role == bad),
3033 "role `{bad}` should be rejected, got {err:?}"
3034 );
3035 }
3036 }
3037
3038 #[test]
3039 fn leaf_role_roundtrips_without_required_provider_model() {
3040 let leaf = LeafSpec {
3041 id: "scout-1".to_string(),
3042 prompt: "Investigate #4090. Read-only.".to_string(),
3043 agent_type: AgentType::Explore,
3044 role: Some("scout".to_string()),
3045 profile: None,
3046 mode: TaskMode::ReadOnly,
3047 isolation: IsolationMode::Shared,
3048 file_scope: Vec::new(),
3049 depends_on_results: Vec::new(),
3050 budget: BudgetSpec::default(),
3051 permissions: PermissionSpec::default(),
3052 model_policy: ModelPolicy::default(),
3053 };
3054 let json = serde_json::to_string(&leaf).expect("serialize");
3055 assert!(json.contains("\"role\":\"scout\""));
3056 let parsed: LeafSpec = serde_json::from_str(&json).expect("parse");
3057 assert_eq!(parsed.role.as_deref(), Some("scout"));
3058 assert_eq!(parsed.model_policy.provider, None);
3060 assert_eq!(parsed.model_policy.model, None);
3061 assert_eq!(parsed.model_policy, ModelPolicy::default());
3062 }
3063
3064 #[test]
3065 fn leaf_profile_token_rule_rejects_invalid_names() {
3066 for bad in ["", "has space", "quote\"y", "role=reviewer", "back`tick"] {
3067 let mut leaf = match leaf_node("scan") {
3068 WorkflowNode::Leaf(leaf) => leaf,
3069 _ => unreachable!("leaf helper returns a leaf"),
3070 };
3071 leaf.profile = Some(bad.to_string());
3072 let workflow = workflow_spec(vec![WorkflowNode::Leaf(leaf)]);
3073
3074 let err = MockWorkflowExecutor::new()
3075 .run(&workflow)
3076 .expect_err("invalid profile token should fail validation");
3077
3078 assert!(
3079 matches!(&err, WorkflowExecutionError::InvalidLeafProfile { profile, .. } if profile == bad),
3080 "profile `{bad}` should be rejected, got {err:?}"
3081 );
3082 }
3083
3084 let mut leaf = match leaf_node("scan") {
3085 WorkflowNode::Leaf(leaf) => leaf,
3086 _ => unreachable!("leaf helper returns a leaf"),
3087 };
3088 leaf.profile = Some("reviewer".to_string());
3089 let workflow = workflow_spec(vec![WorkflowNode::Leaf(leaf)]);
3090 MockWorkflowExecutor::new()
3091 .run(&workflow)
3092 .expect("valid profile token should pass validation");
3093 }
3094
3095 #[test]
3096 fn mock_executor_aggregates_leaf_usage() {
3097 let workflow = workflow_spec(vec![WorkflowNode::BranchSet(BranchSpec {
3098 id: "discover".to_string(),
3099 description: None,
3100 parallel: true,
3101 budget: BudgetSpec::default(),
3102 permissions: PermissionSpec::default(),
3103 model_policy: ModelPolicy::default(),
3104 children: vec![leaf_node("scan-readme"), leaf_node("scan-tests")],
3105 })]);
3106
3107 let mut executor = MockWorkflowExecutor::new()
3108 .with_leaf_outcome(
3109 "scan-readme",
3110 MockLeafOutcome::succeeded("readme ok").with_usage(WorkflowUsage {
3111 input_tokens: 100,
3112 output_tokens: 25,
3113 cost_microusd: 500,
3114 }),
3115 )
3116 .with_leaf_outcome(
3117 "scan-tests",
3118 MockLeafOutcome::succeeded("tests ok").with_usage(WorkflowUsage {
3119 input_tokens: 50,
3120 output_tokens: 10,
3121 cost_microusd: 250,
3122 }),
3123 );
3124
3125 let execution = executor.run(&workflow).expect("mock workflow should run");
3126
3127 assert_eq!(
3128 execution.usage,
3129 WorkflowUsage {
3130 input_tokens: 150,
3131 output_tokens: 35,
3132 cost_microusd: 750,
3133 }
3134 );
3135 assert_eq!(execution.usage.total_tokens(), 185);
3136 assert_eq!(execution.branch_results[0].usage, execution.usage);
3137 assert_eq!(
3138 execution
3139 .leaf_results
3140 .iter()
3141 .map(|result| result.usage.cost_microusd)
3142 .collect::<Vec<_>>(),
3143 vec![500, 250]
3144 );
3145 }
3146
3147 #[test]
3148 fn mock_executor_aggregates_memo_usage() {
3149 let workflow = workflow_spec(vec![WorkflowNode::BranchSet(BranchSpec {
3150 id: "cache-branches".to_string(),
3151 description: None,
3152 parallel: true,
3153 budget: BudgetSpec::default(),
3154 permissions: PermissionSpec::default(),
3155 model_policy: ModelPolicy::default(),
3156 children: vec![leaf_node("rlm-hit"), leaf_node("rlm-miss")],
3157 })]);
3158
3159 let mut executor = MockWorkflowExecutor::new()
3160 .with_leaf_outcome(
3161 "rlm-hit",
3162 MockLeafOutcome::succeeded("memo hit").with_memo_usage(WorkflowMemoUsage {
3163 armh_hits: 1,
3164 armh_misses: 0,
3165 armh_saved_estimated_tokens: 4096,
3166 provider_prompt_cache_hits: 1,
3167 provider_prompt_cache_misses: 0,
3168 }),
3169 )
3170 .with_leaf_outcome(
3171 "rlm-miss",
3172 MockLeafOutcome::succeeded("memo miss").with_memo_usage(WorkflowMemoUsage {
3173 armh_hits: 0,
3174 armh_misses: 1,
3175 armh_saved_estimated_tokens: 0,
3176 provider_prompt_cache_hits: 0,
3177 provider_prompt_cache_misses: 1,
3178 }),
3179 );
3180
3181 let execution = executor.run(&workflow).expect("mock workflow should run");
3182
3183 assert_eq!(
3184 execution.memo_usage,
3185 WorkflowMemoUsage {
3186 armh_hits: 1,
3187 armh_misses: 1,
3188 armh_saved_estimated_tokens: 4096,
3189 provider_prompt_cache_hits: 1,
3190 provider_prompt_cache_misses: 1,
3191 }
3192 );
3193 assert_eq!(execution.branch_results[0].memo_usage, execution.memo_usage);
3194 assert_eq!(
3195 execution
3196 .leaf_results
3197 .iter()
3198 .map(|result| (result.memo_usage.armh_hits, result.memo_usage.armh_misses))
3199 .collect::<Vec<_>>(),
3200 vec![(1, 0), (0, 1)]
3201 );
3202 }
3203
3204 #[test]
3205 fn mock_executor_marks_cancelled_before_leaf() {
3206 let workflow = workflow_spec(vec![WorkflowNode::BranchSet(BranchSpec {
3207 id: "discover".to_string(),
3208 description: None,
3209 parallel: true,
3210 budget: BudgetSpec::default(),
3211 permissions: PermissionSpec::default(),
3212 model_policy: ModelPolicy::default(),
3213 children: vec![leaf_node("scan-readme"), leaf_node("scan-tests")],
3214 })]);
3215
3216 let mut executor = MockWorkflowExecutor::new().with_cancelled();
3217 let execution = executor.run(&workflow).expect("mock workflow should run");
3218
3219 assert_eq!(execution.status, WorkflowRunStatus::Cancelled);
3220 assert_eq!(execution.leaf_results.len(), 1);
3221 assert_eq!(
3222 execution.leaf_results[0].status,
3223 WorkflowRunStatus::Cancelled
3224 );
3225 assert_eq!(
3226 execution.branch_results[0].status,
3227 WorkflowRunStatus::Cancelled
3228 );
3229 assert_eq!(
3230 control_result(&execution, "discover").status,
3231 WorkflowRunStatus::Cancelled
3232 );
3233 }
3234
3235 #[test]
3236 fn mock_executor_stops_when_global_leaf_budget_is_exhausted() {
3237 let workflow = workflow_spec(vec![WorkflowNode::BranchSet(BranchSpec {
3238 id: "discover".to_string(),
3239 description: None,
3240 parallel: true,
3241 budget: BudgetSpec::default(),
3242 permissions: PermissionSpec::default(),
3243 model_policy: ModelPolicy::default(),
3244 children: vec![
3245 leaf_node("scan-readme"),
3246 leaf_node("scan-config"),
3247 leaf_node("scan-tests"),
3248 ],
3249 })]);
3250
3251 let mut executor = MockWorkflowExecutor::new().with_max_leaf_steps(1);
3252 let execution = executor.run(&workflow).expect("mock workflow should run");
3253
3254 assert_eq!(execution.status, WorkflowRunStatus::BudgetExceeded);
3255 assert_eq!(
3256 execution
3257 .leaf_results
3258 .iter()
3259 .map(|result| (result.leaf_id.as_str(), result.status))
3260 .collect::<Vec<_>>(),
3261 vec![
3262 ("scan-readme", WorkflowRunStatus::Succeeded),
3263 ("scan-config", WorkflowRunStatus::BudgetExceeded)
3264 ]
3265 );
3266 assert_eq!(
3267 execution.branch_results[0].status,
3268 WorkflowRunStatus::BudgetExceeded
3269 );
3270 }
3271
3272 #[test]
3273 fn mock_executor_honors_zero_step_leaf_budget() {
3274 let workflow = workflow_spec(vec![WorkflowNode::BranchSet(BranchSpec {
3275 id: "verify".to_string(),
3276 description: None,
3277 parallel: false,
3278 budget: BudgetSpec::default(),
3279 permissions: PermissionSpec::default(),
3280 model_policy: ModelPolicy::default(),
3281 children: vec![
3282 leaf_node_with_budget(
3283 "run-tests",
3284 BudgetSpec {
3285 max_steps: Some(0),
3286 timeout_secs: None,
3287 max_parallel: None,
3288 max_tokens: None,
3289 },
3290 ),
3291 leaf_node("summarize"),
3292 ],
3293 })]);
3294
3295 let mut executor = MockWorkflowExecutor::new();
3296 let execution = executor.run(&workflow).expect("mock workflow should run");
3297
3298 assert_eq!(execution.status, WorkflowRunStatus::BudgetExceeded);
3299 assert_eq!(execution.leaf_results.len(), 1);
3300 assert_eq!(
3301 execution.leaf_results[0].status,
3302 WorkflowRunStatus::BudgetExceeded
3303 );
3304 assert!(
3305 execution.leaf_results[0]
3306 .output
3307 .as_deref()
3308 .unwrap_or_default()
3309 .contains("budget exhausted")
3310 );
3311 }
3312
3313 #[test]
3314 fn mock_executor_stops_when_global_token_budget_is_exhausted() {
3315 let workflow = workflow_spec(vec![WorkflowNode::BranchSet(BranchSpec {
3316 id: "discover".to_string(),
3317 description: None,
3318 parallel: true,
3319 budget: BudgetSpec::default(),
3320 permissions: PermissionSpec::default(),
3321 model_policy: ModelPolicy::default(),
3322 children: vec![
3323 leaf_node("scan-readme"),
3324 leaf_node("scan-config"),
3325 leaf_node("scan-tests"),
3326 ],
3327 })]);
3328
3329 let mut executor = MockWorkflowExecutor::new()
3334 .with_max_leaf_tokens(1000)
3335 .with_leaf_outcome(
3336 "scan-readme",
3337 MockLeafOutcome::succeeded("readme done").with_usage(WorkflowUsage {
3338 input_tokens: 300,
3339 output_tokens: 300,
3340 cost_microusd: 0,
3341 }),
3342 )
3343 .with_leaf_outcome(
3344 "scan-config",
3345 MockLeafOutcome::succeeded("config done").with_usage(WorkflowUsage {
3346 input_tokens: 250,
3347 output_tokens: 250,
3348 cost_microusd: 0,
3349 }),
3350 );
3351 let execution = executor.run(&workflow).expect("mock workflow should run");
3352
3353 assert_eq!(execution.status, WorkflowRunStatus::BudgetExceeded);
3354 assert_eq!(execution.leaf_results.len(), 3);
3360 assert_eq!(
3361 execution.leaf_results[0].status,
3362 WorkflowRunStatus::Succeeded
3363 );
3364 assert_eq!(
3365 execution.leaf_results[1].status,
3366 WorkflowRunStatus::Succeeded
3367 );
3368 assert_eq!(
3369 execution.leaf_results[2].status,
3370 WorkflowRunStatus::BudgetExceeded
3371 );
3372 assert_eq!(execution.usage.total_tokens(), 1100);
3373 }
3374
3375 #[test]
3376 fn mock_executor_honors_zero_token_leaf_budget() {
3377 let workflow = workflow_spec(vec![WorkflowNode::BranchSet(BranchSpec {
3378 id: "verify".to_string(),
3379 description: None,
3380 parallel: false,
3381 budget: BudgetSpec::default(),
3382 permissions: PermissionSpec::default(),
3383 model_policy: ModelPolicy::default(),
3384 children: vec![
3385 leaf_node_with_budget(
3386 "run-tests",
3387 BudgetSpec {
3388 max_steps: None,
3389 timeout_secs: None,
3390 max_parallel: None,
3391 max_tokens: Some(0),
3392 },
3393 ),
3394 leaf_node("summarize"),
3395 ],
3396 })]);
3397
3398 let mut executor = MockWorkflowExecutor::new();
3399 let execution = executor.run(&workflow).expect("mock workflow should run");
3400
3401 assert_eq!(execution.status, WorkflowRunStatus::BudgetExceeded);
3402 assert_eq!(execution.leaf_results.len(), 1);
3403 assert_eq!(
3404 execution.leaf_results[0].status,
3405 WorkflowRunStatus::BudgetExceeded
3406 );
3407 assert!(
3408 execution.leaf_results[0]
3409 .output
3410 .as_deref()
3411 .unwrap_or_default()
3412 .contains("token budget exhausted")
3413 );
3414 }
3415
3416 #[test]
3417 fn mock_executor_honors_per_leaf_token_cap() {
3418 let workflow = workflow_spec(vec![WorkflowNode::BranchSet(BranchSpec {
3419 id: "review".to_string(),
3420 description: None,
3421 parallel: false,
3422 budget: BudgetSpec::default(),
3423 permissions: PermissionSpec::default(),
3424 model_policy: ModelPolicy::default(),
3425 children: vec![
3426 leaf_node_with_budget(
3427 "expensive-scan",
3428 BudgetSpec {
3429 max_steps: None,
3430 timeout_secs: None,
3431 max_parallel: None,
3432 max_tokens: Some(500),
3433 },
3434 ),
3435 leaf_node("summarize"),
3436 ],
3437 })]);
3438
3439 let mut executor = MockWorkflowExecutor::new().with_leaf_outcome(
3441 "expensive-scan",
3442 MockLeafOutcome::succeeded("scan done").with_usage(WorkflowUsage {
3443 input_tokens: 500,
3444 output_tokens: 300,
3445 cost_microusd: 0,
3446 }),
3447 );
3448 let execution = executor.run(&workflow).expect("mock workflow should run");
3449
3450 assert_eq!(execution.status, WorkflowRunStatus::BudgetExceeded);
3451 assert_eq!(execution.leaf_results.len(), 1);
3452 assert_eq!(
3453 execution.leaf_results[0].status,
3454 WorkflowRunStatus::BudgetExceeded
3455 );
3456 assert!(
3457 execution.leaf_results[0]
3458 .output
3459 .as_deref()
3460 .unwrap_or_default()
3461 .contains("token budget exhausted")
3462 );
3463 }
3464
3465 #[test]
3466 fn budget_spec_serializes_max_tokens() {
3467 let budget = BudgetSpec {
3468 max_steps: Some(10),
3469 timeout_secs: Some(600),
3470 max_parallel: Some(4),
3471 max_tokens: Some(50_000),
3472 };
3473 let json = serde_json::to_string(&budget).expect("serialize budget");
3474 let parsed: BudgetSpec = serde_json::from_str(&json).expect("parse budget");
3475 assert_eq!(parsed, budget);
3476 assert!(json.contains("\"max_tokens\":50000"));
3477
3478 let default_json =
3480 serde_json::to_string(&BudgetSpec::default()).expect("serialize default");
3481 let parsed_default: BudgetSpec =
3482 serde_json::from_str(&default_json).expect("parse default budget");
3483 assert_eq!(parsed_default, BudgetSpec::default());
3484 assert!(parsed_default.max_tokens.is_none());
3485 }
3486
3487 #[test]
3488 fn loop_until_stops_on_pass() {
3489 let workflow = workflow_spec(vec![WorkflowNode::LoopUntil(LoopUntilSpec {
3490 id: "verify".to_string(),
3491 condition: "verification passed".to_string(),
3492 max_iterations: Some(5),
3493 children: vec![leaf_node("run-check")],
3494 })]);
3495
3496 let mut executor =
3497 MockWorkflowExecutor::new().with_predicate_results("verify", vec![false, false, true]);
3498 let execution = executor.run(&workflow).expect("loop should run");
3499
3500 assert_eq!(execution.status, WorkflowRunStatus::Succeeded);
3501 assert_eq!(execution.leaf_results.len(), 3);
3502 assert_eq!(
3503 control_result(&execution, "verify").summary.as_deref(),
3504 Some("loop_until iterations=3")
3505 );
3506 }
3507
3508 #[test]
3509 fn loop_until_honors_max_iters() {
3510 let workflow = workflow_spec(vec![WorkflowNode::LoopUntil(LoopUntilSpec {
3511 id: "verify".to_string(),
3512 condition: "verification passed".to_string(),
3513 max_iterations: Some(2),
3514 children: vec![leaf_node("run-check")],
3515 })]);
3516
3517 let mut executor =
3518 MockWorkflowExecutor::new().with_predicate_results("verify", vec![false, false, true]);
3519 let execution = executor.run(&workflow).expect("loop should run");
3520
3521 assert_eq!(execution.status, WorkflowRunStatus::Failed);
3522 assert_eq!(execution.leaf_results.len(), 2);
3523 assert_eq!(
3524 control_result(&execution, "verify").summary.as_deref(),
3525 Some("loop_until iterations=2")
3526 );
3527 }
3528
3529 #[test]
3530 fn cond_uses_logged_predicate_result() {
3531 let workflow = workflow_spec(vec![WorkflowNode::Cond(CondSpec {
3532 id: "should-fix".to_string(),
3533 condition: "finding requires a patch".to_string(),
3534 then_nodes: vec![leaf_node("patch")],
3535 else_nodes: vec![leaf_node("report-only")],
3536 })]);
3537
3538 let mut executor =
3539 MockWorkflowExecutor::new().with_predicate_results("should-fix", vec![true]);
3540 let execution = executor.run(&workflow).expect("cond should run");
3541
3542 assert_eq!(
3543 execution
3544 .leaf_results
3545 .iter()
3546 .map(|result| result.leaf_id.as_str())
3547 .collect::<Vec<_>>(),
3548 vec!["patch"]
3549 );
3550 assert_eq!(
3551 control_result(&execution, "should-fix").summary.as_deref(),
3552 Some("predicate_result=true")
3553 );
3554 }
3555
3556 #[test]
3557 fn expand_respects_max_children() {
3558 let workflow = workflow_spec(vec![WorkflowNode::Expand(ExpandSpec {
3559 id: "split".to_string(),
3560 source: "plan".to_string(),
3561 max_children: Some(2),
3562 template: None,
3563 })]);
3564
3565 let generated = vec![leaf_node("first"), leaf_node("second"), leaf_node("third")];
3566 let mut executor = MockWorkflowExecutor::new().with_generated_nodes("split", generated);
3567 let execution = executor.run(&workflow).expect("expand should run");
3568
3569 assert_eq!(
3570 execution
3571 .leaf_results
3572 .iter()
3573 .map(|result| result.leaf_id.as_str())
3574 .collect::<Vec<_>>(),
3575 vec!["first", "second"]
3576 );
3577 assert_eq!(
3578 control_result(&execution, "split").selected_children,
3579 vec!["first", "second"]
3580 );
3581 }
3582
3583 #[test]
3584 fn expand_generated_nodes_validate_before_run() {
3585 let workflow = workflow_spec(vec![WorkflowNode::Expand(ExpandSpec {
3586 id: "split".to_string(),
3587 source: "plan".to_string(),
3588 max_children: None,
3589 template: None,
3590 })]);
3591
3592 let mut executor = MockWorkflowExecutor::new()
3593 .with_generated_nodes("split", vec![invalid_leaf_node("bad")]);
3594 let err = executor
3595 .run(&workflow)
3596 .expect_err("invalid generated leaf should fail before execution");
3597
3598 assert_eq!(
3599 err,
3600 WorkflowExecutionError::EmptyLeafPrompt {
3601 leaf: "bad".to_string()
3602 }
3603 );
3604 }
3605
3606 #[test]
3607 fn workflow_spec_rejects_unknown_leaf_dependency() {
3608 let mut summarize = leaf_node("summarize");
3609 let WorkflowNode::Leaf(spec) = &mut summarize else {
3610 panic!("expected leaf");
3611 };
3612 spec.depends_on_results = vec!["missing-scan".to_string()];
3613 let workflow = workflow_spec(vec![summarize]);
3614
3615 let mut executor = MockWorkflowExecutor::new();
3616 let err = executor
3617 .run(&workflow)
3618 .expect_err("unknown leaf dependency should fail before execution");
3619
3620 assert_eq!(
3621 err,
3622 WorkflowExecutionError::UnknownNodeReference {
3623 node: "summarize".to_string(),
3624 field: "depends_on_results",
3625 reference: "missing-scan".to_string(),
3626 }
3627 );
3628 }
3629
3630 #[test]
3631 fn workflow_spec_rejects_unknown_reduce_input() {
3632 let workflow = workflow_spec(vec![
3633 leaf_node("scan"),
3634 WorkflowNode::Reduce(ReduceSpec {
3635 id: "summarize".to_string(),
3636 inputs: vec!["scan".to_string(), "missing-review".to_string()],
3637 prompt: "Summarize safe fixes".to_string(),
3638 model_policy: ModelPolicy::default(),
3639 }),
3640 ]);
3641
3642 let mut executor = MockWorkflowExecutor::new();
3643 let err = executor
3644 .run(&workflow)
3645 .expect_err("unknown reduce input should fail before execution");
3646
3647 assert_eq!(
3648 err,
3649 WorkflowExecutionError::UnknownNodeReference {
3650 node: "summarize".to_string(),
3651 field: "inputs",
3652 reference: "missing-review".to_string(),
3653 }
3654 );
3655 }
3656
3657 #[test]
3658 fn workflow_spec_rejects_unknown_teacher_candidate() {
3659 let workflow = workflow_spec(vec![
3660 leaf_node("candidate-a"),
3661 WorkflowNode::TeacherReview(TeacherReviewSpec {
3662 id: "teacher-review".to_string(),
3663 candidates: vec!["candidate-a".to_string(), "candidate-b".to_string()],
3664 promotion_policy: PromotionPolicy::default(),
3665 }),
3666 ]);
3667
3668 let mut executor = MockWorkflowExecutor::new();
3669 let err = executor
3670 .run(&workflow)
3671 .expect_err("unknown teacher candidate should fail before execution");
3672
3673 assert_eq!(
3674 err,
3675 WorkflowExecutionError::UnknownNodeReference {
3676 node: "teacher-review".to_string(),
3677 field: "candidates",
3678 reference: "candidate-b".to_string(),
3679 }
3680 );
3681 }
3682
3683 #[test]
3684 fn teacher_candidate_serialization() {
3685 let candidate = TeacherCandidate {
3686 candidate_id: "teacher-review:branch-a".to_string(),
3687 kind: TeacherCandidateKind::WorkflowRecipe,
3688 status: TeacherCandidateStatus::Proposed,
3689 source_node_id: "branch-a".to_string(),
3690 source_branch_id: Some("branch-a".to_string()),
3691 summary: "Winning branch found a reusable workflow recipe.".to_string(),
3692 evidence: vec![
3693 "status=Succeeded".to_string(),
3694 "tokens=42, cost_microusd=7".to_string(),
3695 ],
3696 replay_results: vec![StudentReplayResult {
3697 trace_id: "trace-a".to_string(),
3698 candidate_id: "teacher-review:branch-a".to_string(),
3699 baseline: StudentReplayMetrics {
3700 score: 70,
3701 cost_microusd: 10,
3702 },
3703 candidate: StudentReplayMetrics {
3704 score: 74,
3705 cost_microusd: 12,
3706 },
3707 required_tests: vec![StudentReplayTestResult {
3708 name: "cargo test -p codewhale-workflow".to_string(),
3709 passed: true,
3710 }],
3711 policy_violations: Vec::new(),
3712 stale: false,
3713 notes: Some("offline replay improved the constrained student".to_string()),
3714 }],
3715 };
3716
3717 let json = serde_json::to_string(&candidate).expect("serialize teacher candidate");
3718
3719 assert!(json.contains("\"kind\":\"workflow_recipe\""));
3720 assert!(json.contains("\"status\":\"proposed\""));
3721 assert!(json.contains("\"replay_results\""));
3722 let parsed: TeacherCandidate =
3723 serde_json::from_str(&json).expect("parse teacher candidate");
3724 assert_eq!(parsed, candidate);
3725 }
3726
3727 #[test]
3728 fn teacher_review_produces_candidate_from_trace() {
3729 let review = TeacherReviewSpec {
3730 id: "teacher-review".to_string(),
3731 candidates: vec!["winning-branch".to_string()],
3732 promotion_policy: PromotionPolicy::default(),
3733 };
3734 let execution = WorkflowExecution {
3735 branch_results: vec![BranchResult {
3736 branch_id: "winning-branch".to_string(),
3737 task_id: "winning-branch".to_string(),
3738 status: WorkflowRunStatus::Succeeded,
3739 usage: WorkflowUsage {
3740 input_tokens: 30,
3741 output_tokens: 12,
3742 cost_microusd: 7,
3743 },
3744 memo_usage: WorkflowMemoUsage::default(),
3745 artifacts: vec!["trace://branches/winning-branch".to_string()],
3746 notes: Some("branch produced a minimal verified patch".to_string()),
3747 }],
3748 ..WorkflowExecution::default()
3749 };
3750
3751 let report = TeacherReviewReport::from_execution(&review, &execution);
3752
3753 assert_eq!(report.review_node_id, "teacher-review");
3754 assert_eq!(report.candidates.len(), 1);
3755 assert_eq!(
3756 report.candidates[0].kind,
3757 TeacherCandidateKind::WorkflowRecipe
3758 );
3759 assert_eq!(
3760 report.candidates[0].status,
3761 TeacherCandidateStatus::Proposed
3762 );
3763 assert!(
3764 report.candidates[0]
3765 .evidence
3766 .iter()
3767 .any(|line| line.contains("tokens=42"))
3768 );
3769 }
3770
3771 #[test]
3772 fn failed_leaf_becomes_regression_test_candidate() {
3773 let review = TeacherReviewSpec {
3774 id: "teacher-review".to_string(),
3775 candidates: vec!["verify-failure".to_string()],
3776 promotion_policy: PromotionPolicy::default(),
3777 };
3778 let execution = WorkflowExecution {
3779 leaf_results: vec![LeafResult {
3780 leaf_id: "verify-failure".to_string(),
3781 task_id: "verify-failure".to_string(),
3782 role: None,
3783 profile: None,
3784 status: WorkflowRunStatus::Failed,
3785 usage: WorkflowUsage::default(),
3786 memo_usage: WorkflowMemoUsage::default(),
3787 output: Some("cargo test failed with a replay mismatch".to_string()),
3788 artifacts: Vec::new(),
3789 schema_error: None,
3790 }],
3791 ..WorkflowExecution::default()
3792 };
3793
3794 let candidates = teacher_candidates_from_execution(&review, &execution);
3795
3796 assert_eq!(candidates.len(), 1);
3797 assert_eq!(candidates[0].kind, TeacherCandidateKind::RegressionTest);
3798 assert_eq!(candidates[0].status, TeacherCandidateStatus::Proposed);
3799 assert!(
3800 candidates[0]
3801 .evidence
3802 .iter()
3803 .any(|line| { line.contains("cargo test failed with a replay mismatch") })
3804 );
3805 }
3806
3807 #[test]
3808 fn student_replay_promotes_only_on_delta() {
3809 let gate = PromotionGate {
3810 min_score_delta: 3,
3811 max_cost_delta_microusd: Some(25),
3812 ..PromotionGate::default()
3813 };
3814 let replay = StudentReplayResult {
3815 trace_id: "trace-a".to_string(),
3816 candidate_id: "teacher-review:branch-a".to_string(),
3817 baseline: StudentReplayMetrics {
3818 score: 80,
3819 cost_microusd: 100,
3820 },
3821 candidate: StudentReplayMetrics {
3822 score: 84,
3823 cost_microusd: 120,
3824 },
3825 required_tests: vec![StudentReplayTestResult {
3826 name: "workflow replay".to_string(),
3827 passed: true,
3828 }],
3829 policy_violations: Vec::new(),
3830 stale: false,
3831 notes: None,
3832 };
3833
3834 let promoted = gate.evaluate_replay("teacher-review:branch-a", &replay);
3835 assert!(promoted.promoted());
3836 assert_eq!(promoted.status, TeacherCandidateStatus::Promoted);
3837 assert_eq!(promoted.score_delta, 4);
3838
3839 let weak_replay = StudentReplayResult {
3840 candidate: StudentReplayMetrics {
3841 score: 82,
3842 cost_microusd: 120,
3843 },
3844 ..replay
3845 };
3846 let rejected = gate.evaluate_replay("teacher-review:branch-a", &weak_replay);
3847 assert!(!rejected.promoted());
3848 assert_eq!(rejected.status, TeacherCandidateStatus::Rejected);
3849 assert!(
3850 rejected
3851 .reasons
3852 .iter()
3853 .any(|reason| reason.contains("below required 3"))
3854 );
3855 }
3856
3857 #[test]
3858 fn promotion_gate_rejects_stale_policy_cost_and_failed_tests() {
3859 let gate = PromotionGate {
3860 min_score_delta: 1,
3861 max_cost_delta_microusd: Some(10),
3862 ..PromotionGate::default()
3863 };
3864 let replay = StudentReplayResult {
3865 trace_id: "trace-a".to_string(),
3866 candidate_id: "teacher-review:branch-a".to_string(),
3867 baseline: StudentReplayMetrics {
3868 score: 70,
3869 cost_microusd: 10,
3870 },
3871 candidate: StudentReplayMetrics {
3872 score: 90,
3873 cost_microusd: 30,
3874 },
3875 required_tests: vec![StudentReplayTestResult {
3876 name: "required regression".to_string(),
3877 passed: false,
3878 }],
3879 policy_violations: vec!["writes outside file scope".to_string()],
3880 stale: true,
3881 notes: None,
3882 };
3883
3884 let decision = gate.evaluate_replay("teacher-review:branch-a", &replay);
3885
3886 assert_eq!(decision.status, TeacherCandidateStatus::Rejected);
3887 assert!(
3888 decision
3889 .reasons
3890 .iter()
3891 .any(|reason| { reason.contains("cost delta 20 exceeds allowed 10") })
3892 );
3893 assert!(
3894 decision
3895 .reasons
3896 .iter()
3897 .any(|reason| { reason.contains("required test `required regression` failed") })
3898 );
3899 assert!(
3900 decision
3901 .reasons
3902 .iter()
3903 .any(|reason| { reason.contains("policy violation: writes outside file scope") })
3904 );
3905 assert!(
3906 decision
3907 .reasons
3908 .iter()
3909 .any(|reason| { reason.contains("student replay result is stale") })
3910 );
3911 }
3912
3913 #[test]
3914 fn promotion_gate_requires_recorded_replay_before_candidate_promotion() {
3915 let candidate = TeacherCandidate {
3916 candidate_id: "teacher-review:branch-a".to_string(),
3917 kind: TeacherCandidateKind::WorkflowRecipe,
3918 status: TeacherCandidateStatus::Proposed,
3919 source_node_id: "branch-a".to_string(),
3920 source_branch_id: Some("branch-a".to_string()),
3921 summary: "candidate waits for replay".to_string(),
3922 evidence: Vec::new(),
3923 replay_results: Vec::new(),
3924 };
3925
3926 let decision = PromotionGate::default().evaluate_candidate(&candidate);
3927
3928 assert_eq!(decision.status, TeacherCandidateStatus::Rejected);
3929 assert_eq!(
3930 decision.reasons,
3931 vec!["no student replay result recorded".to_string()]
3932 );
3933 }
3934
3935 #[test]
3936 fn tournament_selects_passing_minimal_branch() {
3937 let tournament = BranchTournament { min_score: 60 };
3938 let candidates = vec![
3939 candidate(
3940 "expensive-pass",
3941 WorkflowRunStatus::Succeeded,
3942 90,
3943 90,
3944 "quality",
3945 ),
3946 candidate("failed-cheap", WorkflowRunStatus::Failed, 100, 1, "broken"),
3947 candidate(
3948 "cheap-pass",
3949 WorkflowRunStatus::Succeeded,
3950 70,
3951 10,
3952 "minimal",
3953 ),
3954 candidate("too-low", WorkflowRunStatus::Succeeded, 40, 2, "weak"),
3955 ];
3956
3957 let selected = tournament
3958 .select(&candidates)
3959 .expect("one passing branch should be selected");
3960
3961 assert_eq!(selected.branch_id, "cheap-pass");
3962 }
3963
3964 #[test]
3965 fn pareto_frontier_keeps_diverse_candidates() {
3966 let frontier = ParetoFrontier { max_items: 4 };
3967 let candidates = vec![
3968 candidate("quality", WorkflowRunStatus::Succeeded, 95, 100, "quality"),
3969 candidate("minimal", WorkflowRunStatus::Succeeded, 70, 10, "small"),
3970 candidate("dominated", WorkflowRunStatus::Succeeded, 60, 40, "middle"),
3971 candidate("failed", WorkflowRunStatus::Failed, 100, 1, "broken"),
3972 ];
3973
3974 let selected = frontier.select(&candidates);
3975
3976 assert_eq!(
3977 selected
3978 .iter()
3979 .map(|candidate| candidate.branch_id.as_str())
3980 .collect::<Vec<_>>(),
3981 vec!["quality", "minimal"]
3982 );
3983 assert_eq!(
3984 selected
3985 .iter()
3986 .filter_map(|candidate| candidate.diversity_key.as_deref())
3987 .collect::<Vec<_>>(),
3988 vec!["quality", "small"]
3989 );
3990 }
3991}