1use std::fs;
2use std::path::PathBuf;
3use std::time::{Duration, Instant};
4
5use anyhow::{Context, Result};
6use async_trait::async_trait;
7use serde_json::Value;
8use sha2::{Digest, Sha256};
9use tokio::task::JoinSet;
10use tokio::time::MissedTickBehavior;
11
12use crate::workflow_binding::{
13 BindingContext, LoopContext, resolve_bindings, resolve_bound_string,
14};
15use crate::workflow_definition::{HostExecutorNode, SubagentNode};
16use crate::workflow_orchestrator::OrchestratorAction;
17use crate::workflow_sidecar::write_effect_input_sidecar;
18use crate::{
19 EventDraft, EventLog, RunSnapshotDTO, RunStatus, WorkflowActor, WorkflowDefinition,
20 WorkflowNode, WorkflowOutputRef, decide_next_actions, read_run_snapshot,
21};
22
23#[derive(Debug, Clone)]
24pub struct WorkflowRuntimeContext {
25 pub log: EventLog,
26 pub def: WorkflowDefinition,
27 pub runs_base_dir: PathBuf,
28}
29
30#[derive(Debug, Clone, PartialEq, Eq)]
31pub struct WorkflowDispatchSession {
32 pub session_id: String,
33 pub bot_name: String,
34 pub started_at: u64,
35 pub ended_at: Option<u64>,
36 pub cli_session_id: Option<String>,
37 pub lark_app_id: Option<String>,
38 pub cli_id: Option<String>,
39 pub working_dir: Option<String>,
40 pub web_port: Option<u16>,
41 pub log_path: Option<String>,
42}
43
44#[derive(Debug, Clone, PartialEq, Eq)]
45pub enum WorkflowDispatchOutcome {
46 Succeeded {
47 output: Value,
48 session: Option<WorkflowDispatchSession>,
49 },
50 Failed {
51 error_code: String,
52 error_class: String,
53 error_message: String,
54 session: Option<WorkflowDispatchSession>,
55 },
56 Cancelled {
57 cancel_origin_event_id: String,
58 session: Option<WorkflowDispatchSession>,
59 },
60}
61
62#[derive(Debug, Clone)]
63pub struct WorkflowDispatchRun<'a> {
64 pub run_id: &'a str,
65 pub workflow_id: &'a str,
66 pub revision_id: &'a str,
67 pub activity_id: &'a str,
68 pub attempt_id: &'a str,
69 pub node_id: &'a str,
70}
71
72#[async_trait]
73pub trait WorkflowExecutionHooks {
74 async fn execute_subagent(
75 &mut self,
76 ctx: WorkflowDispatchRun<'_>,
77 node: &SubagentNode,
78 resolved_prompt: String,
79 ) -> Result<WorkflowDispatchOutcome>;
80
81 async fn execute_host_executor(
82 &mut self,
83 ctx: WorkflowDispatchRun<'_>,
84 node: &HostExecutorNode,
85 resolved_input: Value,
86 ) -> Result<WorkflowDispatchOutcome>;
87}
88
89#[derive(Debug, Clone, PartialEq, Eq)]
90pub struct RunTickResult {
91 pub actions: usize,
92 pub snapshot: RunSnapshotDTO,
93}
94
95#[derive(Debug, Clone)]
96struct ScheduledAction {
97 action: OrchestratorAction,
98}
99
100#[derive(Debug, Clone, PartialEq, Eq)]
101pub enum RunLoopStopReason {
102 Terminal,
103 NoProgress,
104 AwaitingWait,
105 MaxTicks,
106}
107
108#[derive(Debug, Clone, PartialEq, Eq)]
109pub struct RunLoopResult {
110 pub reason: RunLoopStopReason,
111 pub ticks: usize,
112 pub last_snapshot: RunSnapshotDTO,
113}
114
115pub async fn run_tick<H: WorkflowExecutionHooks + Clone + Send + 'static>(
116 rt: &mut WorkflowRuntimeContext,
117 hooks: &mut H,
118 max_concurrency: usize,
119) -> Result<RunTickResult> {
120 let snapshot = read_snapshot(rt).await?;
121 if matches!(
122 snapshot.run.status,
123 RunStatus::Succeeded | RunStatus::Failed | RunStatus::Cancelled
124 ) {
125 return Ok(RunTickResult {
126 actions: 0,
127 snapshot,
128 });
129 }
130
131 if snapshot_has_pending_cancel(&snapshot) {
132 return Ok(RunTickResult {
133 actions: 0,
134 snapshot,
135 });
136 }
137
138 let actions = select_tick_actions(
139 decide_next_actions(&snapshot, &rt.def),
140 &rt.def,
141 max_concurrency,
142 );
143 if actions.is_empty() {
144 return Ok(RunTickResult {
145 actions: 0,
146 snapshot,
147 });
148 }
149
150 let mut join_set: JoinSet<Result<()>> = JoinSet::new();
151 for scheduled in actions.into_iter() {
152 let mut rt_clone = rt.clone();
153 let mut hooks_clone = hooks.clone();
154 join_set.spawn(async move {
155 apply_orchestrator_action(&mut rt_clone, &mut hooks_clone, scheduled.action).await
156 });
157 }
158
159 let mut applied = 0usize;
160 let mut cancel_poll = tokio::time::interval(Duration::from_millis(20));
161 cancel_poll.set_missed_tick_behavior(MissedTickBehavior::Skip);
162 let mut cancel_seen = false;
163 let mut cancel_abort_deadline: Option<Instant> = None;
164
165 while !join_set.is_empty() {
166 tokio::select! {
167 result = join_set.join_next() => {
168 let Some(result) = result else {
169 break;
170 };
171 match result {
172 Ok(Ok(())) => {
173 applied += 1;
174 let snapshot = read_snapshot(rt).await?;
175 if snapshot_has_pending_cancel(&snapshot)
176 || matches!(
177 snapshot.run.status,
178 RunStatus::Succeeded | RunStatus::Failed | RunStatus::Cancelled
179 )
180 {
181 if snapshot_has_pending_cancel(&snapshot) && !cancel_seen {
182 cancel_seen = true;
183 cancel_abort_deadline =
184 Some(Instant::now() + Duration::from_millis(120));
185 }
186 if matches!(
187 snapshot.run.status,
188 RunStatus::Succeeded | RunStatus::Failed | RunStatus::Cancelled
189 ) {
190 join_set.abort_all();
191 return Ok(RunTickResult {
192 actions: applied,
193 snapshot,
194 });
195 }
196 }
197 }
198 Ok(Err(err)) => {
199 if cancel_seen {
200 continue;
201 }
202 join_set.abort_all();
203 return Err(err);
204 }
205 Err(err) => {
206 if cancel_seen {
207 continue;
208 }
209 join_set.abort_all();
210 return Err(anyhow::anyhow!(err));
211 }
212 }
213 }
214 _ = cancel_poll.tick(), if !cancel_seen => {
215 let snapshot = read_snapshot(rt).await?;
216 if snapshot_has_pending_cancel(&snapshot) {
217 cancel_seen = true;
218 cancel_abort_deadline = Some(Instant::now() + Duration::from_millis(120));
219 join_set.abort_all();
220 }
221 }
222 _ = tokio::time::sleep(Duration::from_millis(20)), if cancel_seen => {
223 if let Some(deadline) = cancel_abort_deadline {
224 if Instant::now() >= deadline {
225 join_set.abort_all();
226 }
227 }
228 }
229 }
230 }
231
232 let snapshot = read_snapshot(rt).await?;
233 Ok(RunTickResult {
234 actions: applied,
235 snapshot,
236 })
237}
238
239pub async fn run_loop<H: WorkflowExecutionHooks + Clone + Send + 'static>(
240 rt: &mut WorkflowRuntimeContext,
241 hooks: &mut H,
242 max_ticks: usize,
243 max_concurrency: usize,
244) -> Result<RunLoopResult> {
245 let mut ticks = 0usize;
246 loop {
247 if ticks >= max_ticks {
248 let snapshot = read_snapshot(rt).await?;
249 return Ok(RunLoopResult {
250 reason: RunLoopStopReason::MaxTicks,
251 ticks,
252 last_snapshot: snapshot,
253 });
254 }
255
256 check_pending_cancels(rt).await?;
257
258 let tick = run_tick(rt, hooks, max_concurrency).await?;
259 ticks += 1;
260 if tick.snapshot.run.status == RunStatus::Succeeded
261 || tick.snapshot.run.status == RunStatus::Failed
262 || tick.snapshot.run.status == RunStatus::Cancelled
263 {
264 return Ok(RunLoopResult {
265 reason: RunLoopStopReason::Terminal,
266 ticks,
267 last_snapshot: tick.snapshot,
268 });
269 }
270 if tick.actions == 0 {
271 let has_waits = !tick.snapshot.dangling.waits.is_empty()
272 && tick
273 .snapshot
274 .dangling
275 .waits
276 .iter()
277 .any(|w| !tick.snapshot.dangling.cancels.contains(w));
278 let reason = if has_waits {
279 RunLoopStopReason::AwaitingWait
280 } else {
281 RunLoopStopReason::NoProgress
282 };
283 return Ok(RunLoopResult {
284 reason,
285 ticks,
286 last_snapshot: tick.snapshot,
287 });
288 }
289 }
290}
291
292async fn check_pending_cancels(rt: &mut WorkflowRuntimeContext) -> Result<()> {
293 let _events = rt.log.read_all()?;
294 let snapshot = read_snapshot(rt).await?;
295 for activity_id in &snapshot.dangling.cancels {
296 if !snapshot.dangling.activities.contains(activity_id) {
297 let attempt_id = snapshot
298 .activities
299 .iter()
300 .find(|a| &a.activity_id == activity_id)
301 .and_then(|a| a.current_attempt_id.clone())
302 .unwrap_or_else(|| format!("{}-attempt-1", activity_id));
303 let _ = crate::complete_activity_cancel(
304 &mut rt.log,
305 crate::CompleteActivityCancelInput {
306 activity_id: activity_id.clone(),
307 attempt_id,
308 cancel_origin_event_id: String::new(),
309 },
310 WorkflowActor::Scheduler,
311 );
312 }
313 }
314
315 if let Some(ref intent) = snapshot.run.cancelled_run_intent {
316 if snapshot.run.status != RunStatus::Cancelled {
317 let _ = crate::complete_run_cancel(
318 &mut rt.log,
319 crate::CompleteRunCancelInput {
320 cancel_origin_event_id: intent.cancel_origin_event_id.clone(),
321 },
322 WorkflowActor::Scheduler,
323 );
324 }
325 }
326
327 if !snapshot.run.cancelled_node_intents.is_empty() {
328 for (node_id, intent) in &snapshot.run.cancelled_node_intents {
329 let _ = crate::complete_node_cancel(
330 &mut rt.log,
331 crate::CompleteNodeCancelInput {
332 node_id: node_id.clone(),
333 cancel_origin_event_id: intent.cancel_origin_event_id.clone(),
334 },
335 WorkflowActor::Scheduler,
336 );
337 }
338 }
339
340 Ok(())
341}
342
343fn snapshot_has_pending_cancel(snapshot: &RunSnapshotDTO) -> bool {
344 snapshot.run.cancelled_run_intent.is_some() || !snapshot.run.cancelled_node_intents.is_empty()
345}
346
347fn select_tick_actions(
348 actions: Vec<OrchestratorAction>,
349 def: &WorkflowDefinition,
350 max_concurrency: usize,
351) -> Vec<ScheduledAction> {
352 let limit = max_concurrency.max(1);
353 let mut selected = Vec::new();
354 let mut seen = std::collections::HashSet::new();
355 for action in actions {
356 let serialization_key = action_serialization_key(def, &action);
357 if seen.insert(serialization_key.clone()) {
358 selected.push(ScheduledAction { action });
359 if selected.len() >= limit {
360 break;
361 }
362 }
363 }
364 selected
365}
366
367fn action_serialization_key(_def: &WorkflowDefinition, action: &OrchestratorAction) -> String {
368 match action {
369 OrchestratorAction::DispatchWork { node_id, node, .. } => {
370 let bot_key = match node {
371 WorkflowNode::Subagent(node) => Some(format!("bot:{}", node.bot)),
372 WorkflowNode::HostExecutor(node) => Some(format!("executor:{}", node.executor)),
373 WorkflowNode::Loop(_) | WorkflowNode::Decision(_) => None,
374 };
375 bot_key.unwrap_or_else(|| format!("node:{node_id}"))
376 }
377 OrchestratorAction::DispatchGate { node_id, .. } => format!("gate:{node_id}"),
378 OrchestratorAction::CompleteNodeSucceeded { node_id, .. }
379 | OrchestratorAction::CompleteNodeFailed { node_id, .. } => {
380 format!("node:{node_id}")
381 }
382 OrchestratorAction::CompleteRunSucceeded { sink_node_id, .. } => {
383 format!("run:{sink_node_id}:succeeded")
384 }
385 OrchestratorAction::CompleteRunFailed { failed_node_id } => {
386 format!("run:{failed_node_id}:failed")
387 }
388 }
389}
390
391pub async fn dispatch_gate(
392 rt: &mut WorkflowRuntimeContext,
393 action: &crate::OrchestratorAction,
394) -> Result<()> {
395 match action {
396 OrchestratorAction::DispatchGate {
397 node_id,
398 activity_id,
399 human_gate,
400 } => {
401 let attempt_id = gate_attempt_id(activity_id);
402 let input_ref = write_json_blob(
403 &mut rt.log,
404 serde_json::json!({
405 "kind": "human-gate",
406 "prompt": human_gate.prompt,
407 "approvers": human_gate.approvers,
408 }),
409 )?;
410 rt.log.append(EventDraft {
411 event_type: "attemptCreated".to_string(),
412 actor: WorkflowActor::Scheduler,
413 payload: serde_json::json!({
414 "nodeId": node_id,
415 "activityId": activity_id,
416 "attemptId": attempt_id,
417 "attemptNumber": 1,
418 "inputRef": input_ref,
419 }),
420 timestamp: None,
421 payload_hash: None,
422 })?;
423
424 let snap = read_snapshot(rt).await?;
425 let ctx = BindingContext {
426 snapshot: &snap,
427 def: &rt.def,
428 run_dir: &rt.log.run_dir,
429 loop_context: None,
430 };
431 let resolved_prompt = resolve_bound_string(&human_gate.prompt, &ctx).await?;
432 let prompt_field = split_prompt(&mut rt.log, &resolved_prompt)?;
433 let _ = crate::create_wait(
434 &mut rt.log,
435 crate::CreateWaitInput {
436 activity_id: activity_id.clone(),
437 attempt_id,
438 node_id: node_id.clone(),
439 wait_kind: crate::WaitKind::HumanGate,
440 deadline_at: human_gate.deadline_ms.map(|ms| now_ms() + ms),
441 prompt: prompt_field.prompt,
442 prompt_ref: prompt_field.prompt_ref,
443 prompt_preview: prompt_field.prompt_preview,
444 approvers: human_gate.approvers.clone(),
445 on_timeout: human_gate.on_timeout.as_deref().map(|v| match v {
446 "success" => crate::WaitOnTimeout::Success,
447 _ => crate::WaitOnTimeout::Fail,
448 }),
449 },
450 )
451 .await?;
452 Ok(())
453 }
454 _ => anyhow::bail!("dispatch_gate called with non-gate action"),
455 }
456}
457
458pub async fn dispatch_work<H: WorkflowExecutionHooks>(
459 rt: &mut WorkflowRuntimeContext,
460 hooks: &mut H,
461 action: &crate::OrchestratorAction,
462) -> Result<WorkflowDispatchOutcome> {
463 match action {
464 OrchestratorAction::DispatchWork {
465 node_id,
466 activity_id,
467 node,
468 } => {
469 let attempt_id = work_attempt_id(activity_id, 1);
470 let input_ref = write_json_blob(
471 &mut rt.log,
472 serde_json::json!({
473 "kind": match node {
474 WorkflowNode::Subagent(_) => "subagent",
475 WorkflowNode::HostExecutor(_) => "hostExecutor",
476 WorkflowNode::Loop(_) => "loop",
477 WorkflowNode::Decision(_) => "decision",
478 },
479 "bot_or_executor": match node {
480 WorkflowNode::Subagent(n) => Value::String(n.bot.clone()),
481 WorkflowNode::HostExecutor(n) => Value::String(n.executor.clone()),
482 _ => Value::Null,
483 },
484 "prompt_or_input": match node {
485 WorkflowNode::Subagent(n) => n.prompt.clone(),
486 WorkflowNode::HostExecutor(n) => n.input.clone(),
487 _ => Value::Null,
488 }
489 }),
490 )?;
491 rt.log.append(EventDraft {
492 event_type: "attemptCreated".to_string(),
493 actor: WorkflowActor::Scheduler,
494 payload: serde_json::json!({
495 "nodeId": node_id,
496 "activityId": activity_id,
497 "attemptId": attempt_id,
498 "attemptNumber": 1,
499 "inputRef": input_ref,
500 }),
501 timestamp: None,
502 payload_hash: None,
503 })?;
504
505 let snap = read_snapshot(rt).await?;
506 let bind_ctx = BindingContext {
507 snapshot: &snap,
508 def: &rt.def,
509 run_dir: &rt.log.run_dir,
510 loop_context: loop_context_from_activity(activity_id),
511 };
512
513 match node {
514 WorkflowNode::Subagent(subagent) => {
515 let resolved_prompt = resolve_bound_string(&subagent.prompt, &bind_ctx).await?;
516 rt.log.append(EventDraft {
517 event_type: "activityRunning".to_string(),
518 actor: WorkflowActor::Scheduler,
519 payload: serde_json::json!({
520 "activityId": activity_id,
521 "attemptId": attempt_id,
522 "leaseId": format!("lease-{}", attempt_id),
523 }),
524 timestamp: None,
525 payload_hash: None,
526 })?;
527 let result = hooks
528 .execute_subagent(
529 WorkflowDispatchRun {
530 run_id: &rt.log.run_id,
531 workflow_id: snap.run.workflow_id.as_deref().unwrap_or(""),
532 revision_id: snap.run.revision_id.as_deref().unwrap_or(""),
533 activity_id,
534 attempt_id: &attempt_id,
535 node_id,
536 },
537 subagent,
538 resolved_prompt,
539 )
540 .await?;
541 settle_work_result(&mut rt.log, activity_id, &attempt_id, result).await
542 }
543 WorkflowNode::HostExecutor(executor) => {
544 let resolved_input = resolve_bindings(&executor.input, &bind_ctx).await?;
545 let parsed_input = resolved_input.clone();
546 let _ = write_effect_input_sidecar(
547 &rt.log,
548 activity_id,
549 &attempt_id,
550 &parsed_input,
551 )
552 .await?;
553 let result = hooks
554 .execute_host_executor(
555 WorkflowDispatchRun {
556 run_id: &rt.log.run_id,
557 workflow_id: snap.run.workflow_id.as_deref().unwrap_or(""),
558 revision_id: snap.run.revision_id.as_deref().unwrap_or(""),
559 activity_id,
560 attempt_id: &attempt_id,
561 node_id,
562 },
563 executor,
564 resolved_input,
565 )
566 .await?;
567 settle_work_result(&mut rt.log, activity_id, &attempt_id, result).await
568 }
569 WorkflowNode::Loop(_) | WorkflowNode::Decision(_) => {
570 anyhow::bail!("dispatch_work received unsupported node type")
571 }
572 }
573 }
574 _ => anyhow::bail!("dispatch_work called with non-work action"),
575 }
576}
577
578pub async fn complete_node_succeeded(
579 log: &mut EventLog,
580 action: &crate::OrchestratorAction,
581) -> Result<()> {
582 if let OrchestratorAction::CompleteNodeSucceeded {
583 node_id,
584 last_activity_id,
585 ..
586 } = action
587 {
588 let _ = log.append(EventDraft {
589 event_type: "nodeSucceeded".to_string(),
590 actor: WorkflowActor::Scheduler,
591 payload: serde_json::json!({
592 "nodeId": node_id,
593 "lastActivityId": last_activity_id,
594 }),
595 timestamp: None,
596 payload_hash: None,
597 })?;
598 Ok(())
599 } else {
600 anyhow::bail!("complete_node_succeeded called with wrong action")
601 }
602}
603
604pub async fn complete_node_failed(
605 log: &mut EventLog,
606 action: &crate::OrchestratorAction,
607) -> Result<()> {
608 if let OrchestratorAction::CompleteNodeFailed {
609 node_id,
610 last_activity_id,
611 error_class,
612 } = action
613 {
614 let _ = log.append(EventDraft {
615 event_type: "nodeFailed".to_string(),
616 actor: WorkflowActor::Scheduler,
617 payload: serde_json::json!({
618 "nodeId": node_id,
619 "lastActivityId": last_activity_id,
620 "errorClass": error_class,
621 }),
622 timestamp: None,
623 payload_hash: None,
624 })?;
625 Ok(())
626 } else {
627 anyhow::bail!("complete_node_failed called with wrong action")
628 }
629}
630
631pub async fn complete_run_succeeded(
632 log: &mut EventLog,
633 action: &crate::OrchestratorAction,
634) -> Result<()> {
635 if let OrchestratorAction::CompleteRunSucceeded { output_ref, .. } = action {
636 let _ = log.append(EventDraft {
637 event_type: "runSucceeded".to_string(),
638 actor: WorkflowActor::Scheduler,
639 payload: serde_json::json!({
640 "outputRef": output_ref,
641 }),
642 timestamp: None,
643 payload_hash: None,
644 })?;
645 Ok(())
646 } else {
647 anyhow::bail!("complete_run_succeeded called with wrong action")
648 }
649}
650
651pub async fn complete_run_failed(
652 log: &mut EventLog,
653 action: &crate::OrchestratorAction,
654) -> Result<()> {
655 if let OrchestratorAction::CompleteRunFailed { failed_node_id } = action {
656 let root_cause_event_id = find_root_cause_event_id(log, failed_node_id).await?;
657 let _ = log.append(EventDraft {
658 event_type: "runFailed".to_string(),
659 actor: WorkflowActor::Scheduler,
660 payload: serde_json::json!({
661 "failedNodeId": failed_node_id,
662 "rootCauseEventId": root_cause_event_id,
663 }),
664 timestamp: None,
665 payload_hash: None,
666 })?;
667 Ok(())
668 } else {
669 anyhow::bail!("complete_run_failed called with wrong action")
670 }
671}
672
673async fn apply_orchestrator_action<H: WorkflowExecutionHooks>(
674 rt: &mut WorkflowRuntimeContext,
675 hooks: &mut H,
676 action: OrchestratorAction,
677) -> Result<()> {
678 match action {
679 OrchestratorAction::DispatchGate { .. } => dispatch_gate(rt, &action).await?,
680 OrchestratorAction::DispatchWork { .. } => {
681 let _ = dispatch_work(rt, hooks, &action).await?;
682 }
683 OrchestratorAction::CompleteNodeSucceeded { .. } => {
684 complete_node_succeeded(&mut rt.log, &action).await?
685 }
686 OrchestratorAction::CompleteNodeFailed { .. } => {
687 complete_node_failed(&mut rt.log, &action).await?
688 }
689 OrchestratorAction::CompleteRunSucceeded { .. } => {
690 complete_run_succeeded(&mut rt.log, &action).await?
691 }
692 OrchestratorAction::CompleteRunFailed { .. } => {
693 complete_run_failed(&mut rt.log, &action).await?
694 }
695 }
696 Ok(())
697}
698
699async fn settle_work_result(
700 log: &mut EventLog,
701 activity_id: &str,
702 attempt_id: &str,
703 result: WorkflowDispatchOutcome,
704) -> Result<WorkflowDispatchOutcome> {
705 match &result {
706 WorkflowDispatchOutcome::Succeeded { output, .. } => {
707 let output_ref = write_json_blob(log, output.clone())?;
708 let _ = log.append(EventDraft {
709 event_type: "activitySucceeded".to_string(),
710 actor: WorkflowActor::Worker,
711 payload: serde_json::json!({
712 "activityId": activity_id,
713 "attemptId": attempt_id,
714 "outputRef": output_ref,
715 }),
716 timestamp: None,
717 payload_hash: None,
718 })?;
719 }
720 WorkflowDispatchOutcome::Failed {
721 error_code,
722 error_class,
723 error_message,
724 ..
725 } => {
726 let _ = log.append(EventDraft {
727 event_type: "activityFailed".to_string(),
728 actor: WorkflowActor::Worker,
729 payload: serde_json::json!({
730 "activityId": activity_id,
731 "attemptId": attempt_id,
732 "error": {
733 "errorCode": error_code,
734 "errorClass": error_class,
735 "errorMessage": error_message,
736 }
737 }),
738 timestamp: None,
739 payload_hash: None,
740 })?;
741 }
742 WorkflowDispatchOutcome::Cancelled {
743 cancel_origin_event_id,
744 ..
745 } => {
746 let _ = log.append(EventDraft {
747 event_type: "activityCanceled".to_string(),
748 actor: WorkflowActor::Worker,
749 payload: serde_json::json!({
750 "activityId": activity_id,
751 "attemptId": attempt_id,
752 "cancelOriginEventId": cancel_origin_event_id,
753 }),
754 timestamp: None,
755 payload_hash: None,
756 })?;
757 }
758 }
759 Ok(result)
760}
761
762async fn find_root_cause_event_id(log: &EventLog, node_id: &str) -> Result<String> {
763 let events = log.read_all()?;
764 let mut node_failed_event_id: Option<String> = None;
765 let mut activity_failed_event_id: Option<String> = None;
766 let mut loop_finished_event_id: Option<String> = None;
767 let mut node_activities = std::collections::BTreeSet::new();
768 for e in &events {
769 match e.event_type.as_str() {
770 "attemptCreated" => {
771 if e.payload.get("nodeId").and_then(Value::as_str) == Some(node_id)
772 && let Some(activity_id) = e.payload.get("activityId").and_then(Value::as_str)
773 {
774 node_activities.insert(activity_id.to_string());
775 }
776 }
777 "activityFailed" => {
778 if let Some(activity_id) = e.payload.get("activityId").and_then(Value::as_str)
779 && node_activities.contains(activity_id)
780 {
781 activity_failed_event_id = Some(e.event_id.clone());
782 }
783 }
784 "nodeFailed" => {
785 if e.payload.get("nodeId").and_then(Value::as_str) == Some(node_id) {
786 node_failed_event_id = Some(e.event_id.clone());
787 }
788 }
789 "loopFinished" => {
790 if e.payload.get("loopId").and_then(Value::as_str) == Some(node_id)
791 && e.payload.get("resolution").and_then(Value::as_str) != Some("approved")
792 {
793 loop_finished_event_id = Some(e.event_id.clone());
794 }
795 }
796 _ => {}
797 }
798 }
799 Ok(activity_failed_event_id
800 .or(node_failed_event_id)
801 .or(loop_finished_event_id)
802 .unwrap_or_else(|| {
803 events
804 .first()
805 .map(|e| e.event_id.clone())
806 .unwrap_or_default()
807 }))
808}
809
810fn gate_attempt_id(activity_id: &str) -> String {
811 format!("{activity_id}::att-1")
812}
813
814fn work_attempt_id(activity_id: &str, attempt_number: u64) -> String {
815 format!("{activity_id}::att-{attempt_number}")
816}
817
818fn now_ms() -> u64 {
819 std::time::SystemTime::now()
820 .duration_since(std::time::UNIX_EPOCH)
821 .unwrap_or_default()
822 .as_millis() as u64
823}
824
825fn split_prompt(log: &mut EventLog, resolved_prompt: &str) -> Result<PromptField> {
826 if resolved_prompt.len() <= 1024 {
827 return Ok(PromptField {
828 prompt: Some(resolved_prompt.to_string()),
829 prompt_ref: None,
830 prompt_preview: None,
831 });
832 }
833 let prompt_ref = write_json_blob(log, Value::String(resolved_prompt.to_string()))?;
834 Ok(PromptField {
835 prompt: None,
836 prompt_ref: Some(prompt_ref),
837 prompt_preview: Some(make_prompt_preview(resolved_prompt)),
838 })
839}
840
841#[derive(Debug)]
842struct PromptField {
843 prompt: Option<String>,
844 prompt_ref: Option<WorkflowOutputRef>,
845 prompt_preview: Option<String>,
846}
847
848fn make_prompt_preview(full: &str) -> String {
849 const MAX: usize = 480;
850 if full.chars().count() <= MAX {
851 return full.to_string();
852 }
853 let suffix = "…(完整内容见 dashboard)";
854 let budget = MAX.saturating_sub(suffix.chars().count());
855 let mut out = String::new();
856 for ch in full.chars().take(budget) {
857 out.push(ch);
858 }
859 out.push_str(suffix);
860 out
861}
862
863fn write_json_blob(log: &mut EventLog, value: Value) -> Result<WorkflowOutputRef> {
864 let bytes = serde_json::to_vec(&value)?;
865 let hash = sha256_hex(&bytes);
866 let path = PathBuf::from(&log.blob_dir).join(&hash);
867 fs::write(&path, &bytes)?;
868 Ok(WorkflowOutputRef {
869 output_hash: format!("sha256:{hash}"),
870 output_path: path.display().to_string(),
871 output_bytes: bytes.len(),
872 output_schema_version: 1,
873 content_type: Some("application/json".to_string()),
874 })
875}
876
877fn sha256_hex(bytes: &[u8]) -> String {
878 let mut hasher = Sha256::new();
879 hasher.update(bytes);
880 format!("{:x}", hasher.finalize())
881}
882
883fn loop_context_from_activity(activity_id: &str) -> Option<LoopContext<'_>> {
884 let loop_start = activity_id.find("::loop::")?;
885 let after_loop = &activity_id[loop_start + "::loop::".len()..];
886 let iter_end = after_loop.find("::")?;
887 let loop_part = &after_loop[..iter_end];
888 let (loop_id, iteration) = loop_part.rsplit_once('.')?;
889 let iteration = iteration.parse().ok()?;
890 Some(LoopContext { loop_id, iteration })
891}
892
893async fn read_snapshot(rt: &WorkflowRuntimeContext) -> Result<RunSnapshotDTO> {
894 read_run_snapshot(&rt.log.run_dir)
895 .await?
896 .context("workflow runtime requires an existing run snapshot")
897}
898
899#[cfg(test)]
900mod tests {
901 use super::*;
902 use crate::RunChatBinding;
903 use crate::workflow_definition::NodeBase;
904 use crate::workflow_snapshot::NodeStatus;
905 use std::collections::{BTreeMap, HashMap};
906 use std::sync::Arc;
907 use tokio::sync::Mutex;
908
909 #[derive(Clone)]
910 struct FakeHooks;
911
912 #[async_trait]
913 impl WorkflowExecutionHooks for FakeHooks {
914 async fn execute_subagent(
915 &mut self,
916 _ctx: WorkflowDispatchRun<'_>,
917 _node: &SubagentNode,
918 resolved_prompt: String,
919 ) -> Result<WorkflowDispatchOutcome> {
920 Ok(WorkflowDispatchOutcome::Succeeded {
921 output: Value::String(resolved_prompt),
922 session: None,
923 })
924 }
925
926 async fn execute_host_executor(
927 &mut self,
928 _ctx: WorkflowDispatchRun<'_>,
929 _node: &HostExecutorNode,
930 resolved_input: Value,
931 ) -> Result<WorkflowDispatchOutcome> {
932 Ok(WorkflowDispatchOutcome::Succeeded {
933 output: resolved_input,
934 session: None,
935 })
936 }
937 }
938
939 fn temp_run_dir(label: &str) -> PathBuf {
940 std::env::temp_dir().join(format!(
941 "beam-workflow-runtime-{label}-{}-{}",
942 std::process::id(),
943 std::time::SystemTime::now()
944 .duration_since(std::time::UNIX_EPOCH)
945 .unwrap_or_default()
946 .as_nanos()
947 ))
948 }
949
950 fn workflow_def() -> WorkflowDefinition {
951 WorkflowDefinition {
952 workflow_id: "flow-a".to_string(),
953 version: 1,
954 params: None,
955 defaults: None,
956 nodes: BTreeMap::from([(
957 "a".to_string(),
958 WorkflowNode::Subagent(SubagentNode {
959 base: NodeBase {
960 description: None,
961 depends: None,
962 human_gate: None,
963 retry_policy: None,
964 timeout_ms: None,
965 max_output_bytes: None,
966 output_schema: None,
967 unsafe_allow_ungated: None,
968 },
969 bot: "bot-a".to_string(),
970 prompt: Value::String("hello ${params.name}".to_string()),
971 working_dir: None,
972 model_overrides: None,
973 tool_policy: None,
974 }),
975 )]),
976 }
977 }
978
979 #[tokio::test]
980 async fn run_tick_dispatches_simple_workflow() {
981 let run_dir = temp_run_dir("tick");
982 let _ = fs::remove_dir_all(&run_dir);
983 fs::create_dir_all(run_dir.join("blobs")).unwrap();
984 fs::write(run_dir.join("params.json"), r#"{"name":"beam"}"#).unwrap();
985
986 let paths = crate::BeamPaths::from_root(run_dir.clone());
987 let params = BTreeMap::from([(String::from("name"), String::from("beam"))]);
988 let run_id = "run-1";
989 let bootstrap = crate::bootstrap_workflow_run(
990 &paths,
991 crate::BootstrapWorkflowRunInput {
992 run_id,
993 workflow_json: r#"{"workflowId":"flow-a","version":1,"nodes":{"a":{"type":"subagent","bot":"bot-a","prompt":"hello ${params.name}"}}}"#,
994 expected_workflow_id: Some("flow-a"),
995 params: ¶ms,
996 initiator: "cli",
997 chat_binding: Some(RunChatBinding {
998 chat_id: "chat-1".to_string(),
999 lark_app_id: "app-1".to_string(),
1000 }),
1001 },
1002 )
1003 .unwrap();
1004 let mut rt = WorkflowRuntimeContext {
1005 log: EventLog::new(run_id, paths.workflow_runs_dir()).unwrap(),
1006 def: workflow_def(),
1007 runs_base_dir: paths.workflow_runs_dir(),
1008 };
1009 let mut hooks = FakeHooks;
1010 let tick = run_tick(&mut rt, &mut hooks, 1).await.unwrap();
1011 assert!(tick.actions > 0);
1012 assert_eq!(tick.snapshot.run.workflow_id.as_deref(), Some("flow-a"));
1013 assert!(matches!(
1014 tick.snapshot
1015 .nodes
1016 .iter()
1017 .find(|node| node.node_id == "a")
1018 .map(|node| node.status),
1019 Some(NodeStatus::Succeeded | NodeStatus::Running | NodeStatus::Triggered)
1020 ));
1021 let _ = fs::remove_dir_all(&run_dir);
1022 let _ = bootstrap;
1023 }
1024
1025 #[tokio::test]
1026 async fn run_tick_honors_max_concurrency_cap() {
1027 let run_dir = temp_run_dir("cap");
1028 let _ = fs::remove_dir_all(&run_dir);
1029 fs::create_dir_all(run_dir.join("blobs")).unwrap();
1030 let paths = crate::BeamPaths::from_root(run_dir.clone());
1031 let params = BTreeMap::new();
1032 let run_id = "run-cap";
1033 crate::bootstrap_workflow_run(
1034 &paths,
1035 crate::BootstrapWorkflowRunInput {
1036 run_id,
1037 workflow_json: r#"{"workflowId":"flow-cap","version":1,"nodes":{"a":{"type":"subagent","bot":"bot-a","prompt":"one"},"b":{"type":"subagent","bot":"bot-b","prompt":"two"}}}"#,
1038 expected_workflow_id: Some("flow-cap"),
1039 params: ¶ms,
1040 initiator: "cli",
1041 chat_binding: Some(RunChatBinding {
1042 chat_id: "chat-1".to_string(),
1043 lark_app_id: "app-1".to_string(),
1044 }),
1045 },
1046 )
1047 .unwrap();
1048 let mut rt = WorkflowRuntimeContext {
1049 log: EventLog::new(run_id, paths.workflow_runs_dir()).unwrap(),
1050 def: WorkflowDefinition {
1051 workflow_id: "flow-cap".to_string(),
1052 version: 1,
1053 params: None,
1054 defaults: None,
1055 nodes: BTreeMap::from([
1056 (
1057 "a".to_string(),
1058 WorkflowNode::Subagent(SubagentNode {
1059 base: NodeBase {
1060 description: None,
1061 depends: None,
1062 human_gate: None,
1063 retry_policy: None,
1064 timeout_ms: None,
1065 max_output_bytes: None,
1066 output_schema: None,
1067 unsafe_allow_ungated: None,
1068 },
1069 bot: "bot-a".to_string(),
1070 prompt: Value::String("one".to_string()),
1071 working_dir: None,
1072 model_overrides: None,
1073 tool_policy: None,
1074 }),
1075 ),
1076 (
1077 "b".to_string(),
1078 WorkflowNode::Subagent(SubagentNode {
1079 base: NodeBase {
1080 description: None,
1081 depends: None,
1082 human_gate: None,
1083 retry_policy: None,
1084 timeout_ms: None,
1085 max_output_bytes: None,
1086 output_schema: None,
1087 unsafe_allow_ungated: None,
1088 },
1089 bot: "bot-b".to_string(),
1090 prompt: Value::String("two".to_string()),
1091 working_dir: None,
1092 model_overrides: None,
1093 tool_policy: None,
1094 }),
1095 ),
1096 ]),
1097 },
1098 runs_base_dir: paths.workflow_runs_dir(),
1099 };
1100 let mut hooks = FakeHooks;
1101 let tick = run_tick(&mut rt, &mut hooks, 1).await.unwrap();
1102 assert_eq!(tick.actions, 1);
1103 let snapshot = tick.snapshot;
1104 let attempted: Vec<_> = snapshot
1105 .activities
1106 .iter()
1107 .map(|a| a.activity_id.as_str())
1108 .collect();
1109 assert_eq!(attempted.len(), 1);
1110 let _ = fs::remove_dir_all(&run_dir);
1111 }
1112
1113 #[tokio::test]
1114 async fn run_tick_dispatches_multiple_actions_concurrently() {
1115 let run_dir = temp_run_dir("concurrent");
1116 let _ = fs::remove_dir_all(&run_dir);
1117 fs::create_dir_all(run_dir.join("blobs")).unwrap();
1118 let paths = crate::BeamPaths::from_root(run_dir.clone());
1119 let params = BTreeMap::new();
1120 let run_id = "run-concurrent";
1121 crate::bootstrap_workflow_run(
1122 &paths,
1123 crate::BootstrapWorkflowRunInput {
1124 run_id,
1125 workflow_json: r#"{"workflowId":"flow-concurrent","version":1,"nodes":{"a":{"type":"subagent","bot":"bot-a","prompt":"one"},"b":{"type":"subagent","bot":"bot-b","prompt":"two"}}}"#,
1126 expected_workflow_id: Some("flow-concurrent"),
1127 params: ¶ms,
1128 initiator: "cli",
1129 chat_binding: Some(RunChatBinding {
1130 chat_id: "chat-1".to_string(),
1131 lark_app_id: "app-1".to_string(),
1132 }),
1133 },
1134 )
1135 .unwrap();
1136 let mut rt = WorkflowRuntimeContext {
1137 log: EventLog::new(run_id, paths.workflow_runs_dir()).unwrap(),
1138 def: WorkflowDefinition {
1139 workflow_id: "flow-concurrent".to_string(),
1140 version: 1,
1141 params: None,
1142 defaults: None,
1143 nodes: BTreeMap::from([
1144 (
1145 "a".to_string(),
1146 WorkflowNode::Subagent(SubagentNode {
1147 base: NodeBase {
1148 description: None,
1149 depends: None,
1150 human_gate: None,
1151 retry_policy: None,
1152 timeout_ms: None,
1153 max_output_bytes: None,
1154 output_schema: None,
1155 unsafe_allow_ungated: None,
1156 },
1157 bot: "bot-a".to_string(),
1158 prompt: Value::String("one".to_string()),
1159 working_dir: None,
1160 model_overrides: None,
1161 tool_policy: None,
1162 }),
1163 ),
1164 (
1165 "b".to_string(),
1166 WorkflowNode::Subagent(SubagentNode {
1167 base: NodeBase {
1168 description: None,
1169 depends: None,
1170 human_gate: None,
1171 retry_policy: None,
1172 timeout_ms: None,
1173 max_output_bytes: None,
1174 output_schema: None,
1175 unsafe_allow_ungated: None,
1176 },
1177 bot: "bot-b".to_string(),
1178 prompt: Value::String("two".to_string()),
1179 working_dir: None,
1180 model_overrides: None,
1181 tool_policy: None,
1182 }),
1183 ),
1184 ]),
1185 },
1186 runs_base_dir: paths.workflow_runs_dir(),
1187 };
1188
1189 #[derive(Clone)]
1190 struct SleepyHooks;
1191
1192 #[async_trait]
1193 impl WorkflowExecutionHooks for SleepyHooks {
1194 async fn execute_subagent(
1195 &mut self,
1196 _ctx: WorkflowDispatchRun<'_>,
1197 _node: &SubagentNode,
1198 resolved_prompt: String,
1199 ) -> Result<WorkflowDispatchOutcome> {
1200 tokio::time::sleep(std::time::Duration::from_millis(120)).await;
1201 Ok(WorkflowDispatchOutcome::Succeeded {
1202 output: Value::String(resolved_prompt),
1203 session: None,
1204 })
1205 }
1206
1207 async fn execute_host_executor(
1208 &mut self,
1209 _ctx: WorkflowDispatchRun<'_>,
1210 _node: &HostExecutorNode,
1211 resolved_input: Value,
1212 ) -> Result<WorkflowDispatchOutcome> {
1213 Ok(WorkflowDispatchOutcome::Succeeded {
1214 output: resolved_input,
1215 session: None,
1216 })
1217 }
1218 }
1219
1220 let mut hooks = SleepyHooks;
1221 let started = std::time::Instant::now();
1222 let tick = run_tick(&mut rt, &mut hooks, 2).await.unwrap();
1223 let elapsed = started.elapsed();
1224 assert_eq!(tick.actions, 2);
1225 assert!(
1226 elapsed < std::time::Duration::from_millis(220),
1227 "run_tick took {:?}, expected concurrent execution under 220ms",
1228 elapsed
1229 );
1230 let _ = fs::remove_dir_all(&run_dir);
1231 }
1232
1233 #[tokio::test]
1234 async fn run_tick_serializes_actions_for_the_same_bot() {
1235 let run_dir = temp_run_dir("same-bot");
1236 let _ = fs::remove_dir_all(&run_dir);
1237 fs::create_dir_all(run_dir.join("blobs")).unwrap();
1238 let paths = crate::BeamPaths::from_root(run_dir.clone());
1239 let params = BTreeMap::new();
1240 let run_id = "run-same-bot";
1241 crate::bootstrap_workflow_run(
1242 &paths,
1243 crate::BootstrapWorkflowRunInput {
1244 run_id,
1245 workflow_json: r#"{"workflowId":"flow-same-bot","version":1,"nodes":{"a":{"type":"subagent","bot":"bot-shared","prompt":"one"},"b":{"type":"subagent","bot":"bot-shared","prompt":"two"},"c":{"type":"subagent","bot":"bot-other","prompt":"three"}}}"#,
1246 expected_workflow_id: Some("flow-same-bot"),
1247 params: ¶ms,
1248 initiator: "cli",
1249 chat_binding: Some(RunChatBinding {
1250 chat_id: "chat-1".to_string(),
1251 lark_app_id: "app-1".to_string(),
1252 }),
1253 },
1254 )
1255 .unwrap();
1256 let mut rt = WorkflowRuntimeContext {
1257 log: EventLog::new(run_id, paths.workflow_runs_dir()).unwrap(),
1258 def: WorkflowDefinition {
1259 workflow_id: "flow-same-bot".to_string(),
1260 version: 1,
1261 params: None,
1262 defaults: None,
1263 nodes: BTreeMap::from([
1264 (
1265 "a".to_string(),
1266 WorkflowNode::Subagent(SubagentNode {
1267 base: NodeBase {
1268 description: None,
1269 depends: None,
1270 human_gate: None,
1271 retry_policy: None,
1272 timeout_ms: None,
1273 max_output_bytes: None,
1274 output_schema: None,
1275 unsafe_allow_ungated: None,
1276 },
1277 bot: "bot-shared".to_string(),
1278 prompt: Value::String("one".to_string()),
1279 working_dir: None,
1280 model_overrides: None,
1281 tool_policy: None,
1282 }),
1283 ),
1284 (
1285 "b".to_string(),
1286 WorkflowNode::Subagent(SubagentNode {
1287 base: NodeBase {
1288 description: None,
1289 depends: None,
1290 human_gate: None,
1291 retry_policy: None,
1292 timeout_ms: None,
1293 max_output_bytes: None,
1294 output_schema: None,
1295 unsafe_allow_ungated: None,
1296 },
1297 bot: "bot-shared".to_string(),
1298 prompt: Value::String("two".to_string()),
1299 working_dir: None,
1300 model_overrides: None,
1301 tool_policy: None,
1302 }),
1303 ),
1304 (
1305 "c".to_string(),
1306 WorkflowNode::Subagent(SubagentNode {
1307 base: NodeBase {
1308 description: None,
1309 depends: None,
1310 human_gate: None,
1311 retry_policy: None,
1312 timeout_ms: None,
1313 max_output_bytes: None,
1314 output_schema: None,
1315 unsafe_allow_ungated: None,
1316 },
1317 bot: "bot-other".to_string(),
1318 prompt: Value::String("three".to_string()),
1319 working_dir: None,
1320 model_overrides: None,
1321 tool_policy: None,
1322 }),
1323 ),
1324 ]),
1325 },
1326 runs_base_dir: paths.workflow_runs_dir(),
1327 };
1328
1329 #[derive(Clone)]
1330 struct SerializingHooks {
1331 started: Arc<Mutex<Vec<String>>>,
1332 active_by_bot: Arc<Mutex<HashMap<String, usize>>>,
1333 max_active_by_bot: Arc<Mutex<HashMap<String, usize>>>,
1334 }
1335
1336 #[async_trait]
1337 impl WorkflowExecutionHooks for SerializingHooks {
1338 async fn execute_subagent(
1339 &mut self,
1340 _ctx: WorkflowDispatchRun<'_>,
1341 node: &SubagentNode,
1342 resolved_prompt: String,
1343 ) -> Result<WorkflowDispatchOutcome> {
1344 let bot = node.bot.clone();
1345 {
1346 let mut active = self.active_by_bot.lock().await;
1347 let entry = active.entry(bot.clone()).or_insert(0);
1348 *entry += 1;
1349 let mut max_active = self.max_active_by_bot.lock().await;
1350 let max_entry = max_active.entry(bot.clone()).or_insert(0);
1351 if *entry > *max_entry {
1352 *max_entry = *entry;
1353 }
1354 assert!(
1355 *entry <= 1,
1356 "bot {} ran concurrently with itself: {}",
1357 bot,
1358 *entry
1359 );
1360 }
1361 self.started.lock().await.push(resolved_prompt.clone());
1362 tokio::time::sleep(std::time::Duration::from_millis(120)).await;
1363 {
1364 let mut active = self.active_by_bot.lock().await;
1365 if let Some(entry) = active.get_mut(&bot) {
1366 *entry = entry.saturating_sub(1);
1367 }
1368 }
1369 Ok(WorkflowDispatchOutcome::Succeeded {
1370 output: Value::String(resolved_prompt),
1371 session: None,
1372 })
1373 }
1374
1375 async fn execute_host_executor(
1376 &mut self,
1377 _ctx: WorkflowDispatchRun<'_>,
1378 _node: &HostExecutorNode,
1379 resolved_input: Value,
1380 ) -> Result<WorkflowDispatchOutcome> {
1381 Ok(WorkflowDispatchOutcome::Succeeded {
1382 output: resolved_input,
1383 session: None,
1384 })
1385 }
1386 }
1387
1388 let started = Arc::new(Mutex::new(Vec::new()));
1389 let active_by_bot = Arc::new(Mutex::new(HashMap::new()));
1390 let max_active_by_bot = Arc::new(Mutex::new(HashMap::new()));
1391 let mut hooks = SerializingHooks {
1392 started: started.clone(),
1393 active_by_bot: active_by_bot.clone(),
1394 max_active_by_bot: max_active_by_bot.clone(),
1395 };
1396
1397 let started_at = std::time::Instant::now();
1398 let tick = run_tick(&mut rt, &mut hooks, 2).await.unwrap();
1399 let elapsed = started_at.elapsed();
1400
1401 assert_eq!(tick.actions, 2);
1402 let started = started.lock().await.clone();
1403 assert_eq!(started.len(), 2);
1404 assert!(started.contains(&"one".to_string()));
1405 assert!(started.contains(&"three".to_string()));
1406 assert!(!started.contains(&"two".to_string()));
1407 let max_active_by_bot = max_active_by_bot.lock().await;
1408 assert_eq!(max_active_by_bot.get("bot-shared"), Some(&1));
1409 assert_eq!(max_active_by_bot.get("bot-other"), Some(&1));
1410 assert!(
1411 elapsed < std::time::Duration::from_millis(220),
1412 "run_tick took {:?}, expected two distinct bots to run concurrently",
1413 elapsed
1414 );
1415 let _ = fs::remove_dir_all(&run_dir);
1416 }
1417
1418 #[derive(Clone)]
1419 struct CancellingHooks {
1420 run_id: String,
1421 runs_base_dir: PathBuf,
1422 calls: Arc<Mutex<usize>>,
1423 completed: Arc<Mutex<usize>>,
1424 }
1425
1426 #[async_trait]
1427 impl WorkflowExecutionHooks for CancellingHooks {
1428 async fn execute_subagent(
1429 &mut self,
1430 _ctx: WorkflowDispatchRun<'_>,
1431 _node: &SubagentNode,
1432 resolved_prompt: String,
1433 ) -> Result<WorkflowDispatchOutcome> {
1434 let mut calls = self.calls.lock().await;
1435 *calls += 1;
1436 if *calls == 1 {
1437 let run_id = self.run_id.clone();
1438 let runs_base_dir = self.runs_base_dir.clone();
1439 tokio::spawn(async move {
1440 tokio::time::sleep(std::time::Duration::from_millis(1)).await;
1441 let mut log = EventLog::new(run_id.clone(), runs_base_dir).unwrap();
1442 let _ = log.append(EventDraft {
1443 event_type: "cancelRequested".to_string(),
1444 actor: WorkflowActor::Human,
1445 payload: serde_json::json!({
1446 "target": { "kind": "run", "runId": run_id },
1447 "reason": "cancel mid tick",
1448 "by": "tester",
1449 }),
1450 timestamp: None,
1451 payload_hash: None,
1452 });
1453 });
1454 }
1455 tokio::time::sleep(std::time::Duration::from_millis(500)).await;
1456 *self.completed.lock().await += 1;
1457 Ok(WorkflowDispatchOutcome::Succeeded {
1458 output: Value::String(resolved_prompt),
1459 session: None,
1460 })
1461 }
1462
1463 async fn execute_host_executor(
1464 &mut self,
1465 _ctx: WorkflowDispatchRun<'_>,
1466 _node: &HostExecutorNode,
1467 resolved_input: Value,
1468 ) -> Result<WorkflowDispatchOutcome> {
1469 Ok(WorkflowDispatchOutcome::Succeeded {
1470 output: resolved_input,
1471 session: None,
1472 })
1473 }
1474 }
1475
1476 #[tokio::test]
1477 async fn run_tick_stops_between_actions_when_cancel_arrives() {
1478 let run_dir = temp_run_dir("cancel-mid-tick");
1479 let _ = fs::remove_dir_all(&run_dir);
1480 fs::create_dir_all(run_dir.join("blobs")).unwrap();
1481 let paths = crate::BeamPaths::from_root(run_dir.clone());
1482 let run_id = "run-cancel";
1483 crate::bootstrap_workflow_run(
1484 &paths,
1485 crate::BootstrapWorkflowRunInput {
1486 run_id,
1487 workflow_json: r#"{"workflowId":"flow-cancel","version":1,"nodes":{"a":{"type":"subagent","bot":"bot-a","prompt":"one"},"b":{"type":"subagent","bot":"bot-b","prompt":"two"}}}"#,
1488 expected_workflow_id: Some("flow-cancel"),
1489 params: &BTreeMap::new(),
1490 initiator: "cli",
1491 chat_binding: Some(RunChatBinding {
1492 chat_id: "chat-1".to_string(),
1493 lark_app_id: "app-1".to_string(),
1494 }),
1495 },
1496 )
1497 .unwrap();
1498 let mut rt = WorkflowRuntimeContext {
1499 log: EventLog::new(run_id, paths.workflow_runs_dir()).unwrap(),
1500 def: WorkflowDefinition {
1501 workflow_id: "flow-cancel".to_string(),
1502 version: 1,
1503 params: None,
1504 defaults: None,
1505 nodes: BTreeMap::from([
1506 (
1507 "a".to_string(),
1508 WorkflowNode::Subagent(SubagentNode {
1509 base: NodeBase {
1510 description: None,
1511 depends: None,
1512 human_gate: None,
1513 retry_policy: None,
1514 timeout_ms: None,
1515 max_output_bytes: None,
1516 output_schema: None,
1517 unsafe_allow_ungated: None,
1518 },
1519 bot: "bot-a".to_string(),
1520 prompt: Value::String("one".to_string()),
1521 working_dir: None,
1522 model_overrides: None,
1523 tool_policy: None,
1524 }),
1525 ),
1526 (
1527 "b".to_string(),
1528 WorkflowNode::Subagent(SubagentNode {
1529 base: NodeBase {
1530 description: None,
1531 depends: None,
1532 human_gate: None,
1533 retry_policy: None,
1534 timeout_ms: None,
1535 max_output_bytes: None,
1536 output_schema: None,
1537 unsafe_allow_ungated: None,
1538 },
1539 bot: "bot-b".to_string(),
1540 prompt: Value::String("two".to_string()),
1541 working_dir: None,
1542 model_overrides: None,
1543 tool_policy: None,
1544 }),
1545 ),
1546 ]),
1547 },
1548 runs_base_dir: paths.workflow_runs_dir(),
1549 };
1550 let mut hooks = CancellingHooks {
1551 run_id: run_id.to_string(),
1552 runs_base_dir: paths.workflow_runs_dir(),
1553 calls: Arc::new(Mutex::new(0)),
1554 completed: Arc::new(Mutex::new(0)),
1555 };
1556 let started = std::time::Instant::now();
1557 let tick = run_tick(&mut rt, &mut hooks, 2).await.unwrap();
1558 let elapsed = started.elapsed();
1559 assert!(tick.actions < 2);
1560 assert!(tick.snapshot.run.cancelled_run_intent.is_some());
1561 assert!(tick.snapshot.activities.len() >= 1);
1562 assert!(
1563 elapsed < std::time::Duration::from_millis(250),
1564 "run_tick took {:?}, expected cancel to interrupt long-running actions",
1565 elapsed
1566 );
1567 assert_eq!(*hooks.completed.lock().await, 0);
1568 let _ = fs::remove_dir_all(&run_dir);
1569 }
1570
1571 #[tokio::test]
1572 async fn run_tick_skips_dispatch_when_run_cancel_is_pending() {
1573 let run_dir = temp_run_dir("reconcile-cancel");
1574 let _ = fs::remove_dir_all(&run_dir);
1575 fs::create_dir_all(run_dir.join("blobs")).unwrap();
1576 let paths = crate::BeamPaths::from_root(run_dir.clone());
1577 let run_id = "run-reconcile-cancel";
1578 crate::bootstrap_workflow_run(
1579 &paths,
1580 crate::BootstrapWorkflowRunInput {
1581 run_id,
1582 workflow_json: r#"{"workflowId":"flow-reconcile","version":1,"nodes":{"a":{"type":"subagent","bot":"bot-a","prompt":"one"},"b":{"type":"subagent","bot":"bot-b","prompt":"two"}}}"#,
1583 expected_workflow_id: Some("flow-reconcile"),
1584 params: &BTreeMap::new(),
1585 initiator: "cli",
1586 chat_binding: Some(RunChatBinding {
1587 chat_id: "chat-1".to_string(),
1588 lark_app_id: "app-1".to_string(),
1589 }),
1590 },
1591 )
1592 .unwrap();
1593 let mut log = EventLog::new(run_id, paths.workflow_runs_dir()).unwrap();
1594 let _ = crate::request_cancel(
1595 &mut log,
1596 crate::RequestCancelInput {
1597 target: serde_json::json!({
1598 "kind": "run",
1599 "runId": run_id,
1600 }),
1601 reason: "cancel before dispatch".to_string(),
1602 by: "tester".to_string(),
1603 },
1604 WorkflowActor::Human,
1605 )
1606 .await
1607 .unwrap();
1608 let mut rt = WorkflowRuntimeContext {
1609 log: EventLog::new(run_id, paths.workflow_runs_dir()).unwrap(),
1610 def: WorkflowDefinition {
1611 workflow_id: "flow-reconcile".to_string(),
1612 version: 1,
1613 params: None,
1614 defaults: None,
1615 nodes: BTreeMap::from([
1616 (
1617 "a".to_string(),
1618 WorkflowNode::Subagent(SubagentNode {
1619 base: NodeBase {
1620 description: None,
1621 depends: None,
1622 human_gate: None,
1623 retry_policy: None,
1624 timeout_ms: None,
1625 max_output_bytes: None,
1626 output_schema: None,
1627 unsafe_allow_ungated: None,
1628 },
1629 bot: "bot-a".to_string(),
1630 prompt: Value::String("one".to_string()),
1631 working_dir: None,
1632 model_overrides: None,
1633 tool_policy: None,
1634 }),
1635 ),
1636 (
1637 "b".to_string(),
1638 WorkflowNode::Subagent(SubagentNode {
1639 base: NodeBase {
1640 description: None,
1641 depends: None,
1642 human_gate: None,
1643 retry_policy: None,
1644 timeout_ms: None,
1645 max_output_bytes: None,
1646 output_schema: None,
1647 unsafe_allow_ungated: None,
1648 },
1649 bot: "bot-b".to_string(),
1650 prompt: Value::String("two".to_string()),
1651 working_dir: None,
1652 model_overrides: None,
1653 tool_policy: None,
1654 }),
1655 ),
1656 ]),
1657 },
1658 runs_base_dir: paths.workflow_runs_dir(),
1659 };
1660 let mut hooks = FakeHooks;
1661 let tick = run_tick(&mut rt, &mut hooks, 2).await.unwrap();
1662 assert_eq!(tick.actions, 0);
1663 assert!(tick.snapshot.run.cancelled_run_intent.is_some());
1664 let _ = fs::remove_dir_all(&run_dir);
1665 }
1666
1667 #[test]
1668 fn orchestrator_action_is_dispatch_classifies_correctly() {
1669 use crate::workflow_definition::{NodeBase, SubagentNode};
1670 use crate::workflow_orchestrator::OrchestratorAction;
1671 let dispatch = OrchestratorAction::DispatchWork {
1672 node_id: "n1".to_string(),
1673 activity_id: "a1".to_string(),
1674 node: crate::WorkflowNode::Subagent(SubagentNode {
1675 base: NodeBase {
1676 description: None,
1677 depends: None,
1678 human_gate: None,
1679 retry_policy: None,
1680 timeout_ms: None,
1681 max_output_bytes: None,
1682 output_schema: None,
1683 unsafe_allow_ungated: None,
1684 },
1685 bot: "b1".to_string(),
1686 prompt: serde_json::json!("p"),
1687 working_dir: None,
1688 model_overrides: None,
1689 tool_policy: None,
1690 }),
1691 };
1692 assert!(dispatch.is_dispatch());
1693
1694 let settle = OrchestratorAction::CompleteNodeSucceeded {
1695 node_id: "n1".to_string(),
1696 last_activity_id: "a1".to_string(),
1697 output_ref: None,
1698 };
1699 assert!(!settle.is_dispatch());
1700 }
1701
1702 #[tokio::test]
1703 async fn run_loop_stops_when_progress_is_exhausted() {
1704 let run_dir = temp_run_dir("loop");
1705 let _ = fs::remove_dir_all(&run_dir);
1706 fs::create_dir_all(run_dir.join("blobs")).unwrap();
1707 let paths = crate::BeamPaths::from_root(run_dir.clone());
1708 let params = BTreeMap::from([(String::from("name"), String::from("beam"))]);
1709 let run_id = "run-1";
1710 crate::bootstrap_workflow_run(
1711 &paths,
1712 crate::BootstrapWorkflowRunInput {
1713 run_id,
1714 workflow_json: r#"{"workflowId":"flow-a","version":1,"nodes":{"a":{"type":"subagent","bot":"bot-a","prompt":"hello ${params.name}"}}}"#,
1715 expected_workflow_id: Some("flow-a"),
1716 params: ¶ms,
1717 initiator: "cli",
1718 chat_binding: Some(RunChatBinding {
1719 chat_id: "chat-1".to_string(),
1720 lark_app_id: "app-1".to_string(),
1721 }),
1722 },
1723 )
1724 .unwrap();
1725 let mut rt = WorkflowRuntimeContext {
1726 log: EventLog::new(run_id, paths.workflow_runs_dir()).unwrap(),
1727 def: workflow_def(),
1728 runs_base_dir: paths.workflow_runs_dir(),
1729 };
1730 let mut hooks = FakeHooks;
1731 let result = run_loop(&mut rt, &mut hooks, 3, 1).await.unwrap();
1732 assert!(matches!(
1733 result.reason,
1734 RunLoopStopReason::Terminal | RunLoopStopReason::NoProgress
1735 ));
1736 assert!(result.ticks > 0);
1737 let _ = fs::remove_dir_all(&run_dir);
1738 }
1739}