A *named* barrier and a resume boundary. Emits [`WorkflowEvent::PhaseStart`]/[`WorkflowEvent::PhaseEnd`] around the inner combinator. When a store is configured the phase runs through [`execute_steps_parallel_resumable`] keyed by a deterministic `"{root_id}/{index}:{name}"` id, so an interrupted run skips already completed steps; otherwise it is a plain [`parallel`](Self::parallel).
(&self, name: &str, specs: Vec<AgentStepSpec>)
| 235 | /// `"{root_id}/{index}:{name}"` id, so an interrupted run skips already |
| 236 | /// completed steps; otherwise it is a plain [`parallel`](Self::parallel). |
| 237 | pub async fn phase(&self, name: &str, specs: Vec<AgentStepSpec>) -> Vec<StepOutcome> { |
| 238 | let index = self.phase_seq.fetch_add(1, Ordering::SeqCst); |
| 239 | let _ = self.events.send(WorkflowEvent::PhaseStart { |
| 240 | name: name.to_string(), |
| 241 | index, |
| 242 | step_count: specs.len(), |
| 243 | }); |
| 244 | |
| 245 | let out = match &self.store { |
| 246 | Some(store) => { |
| 247 | let workflow_id = format!("{}/{index}:{name}", self.root_id); |
| 248 | execute_steps_parallel_resumable( |
| 249 | Arc::clone(&self.executor), |
| 250 | specs, |
| 251 | &workflow_id, |
| 252 | Arc::clone(store), |
| 253 | self.step_events.clone(), |
| 254 | ) |
| 255 | .await |
| 256 | } |
| 257 | None => self.parallel(specs).await, |
| 258 | }; |
| 259 | |
| 260 | let failed = out.iter().filter(|o| !o.success).count(); |
| 261 | let _ = self.events.send(WorkflowEvent::PhaseEnd { |
| 262 | name: name.to_string(), |
| 263 | index, |
| 264 | succeeded: out.len() - failed, |
| 265 | failed, |
| 266 | }); |
| 267 | |
| 268 | // Surface a budget cap once it is reached so the host can stop issuing |
| 269 | // further phases. Enforcement itself lives in the child loops (via the |
| 270 | // shared budget guard); this is observation only. |
| 271 | if let Some(budget) = &self.budget { |
| 272 | if budget.is_exhausted() { |
| 273 | let snap = budget.snapshot(); |
| 274 | let _ = self.events.send(WorkflowEvent::BudgetExhausted { |
| 275 | resource: "workflow_tokens".to_string(), |
| 276 | reason: format!( |
| 277 | "workflow token budget exhausted ({} / {} tokens)", |
| 278 | snap.consumed_tokens, |
| 279 | snap.limit_tokens.unwrap_or(0) |
| 280 | ), |
| 281 | }); |
| 282 | } |
| 283 | } |
| 284 | out |
| 285 | } |
| 286 | |
| 287 | /// Per-item staged chains with **no barrier between stages** — delegates |
| 288 | /// straight to [`execute_pipeline`]. Item A may be in stage 3 while item B is |