1use std::path::PathBuf;
2use std::sync::{Arc, Condvar, Mutex};
3use std::time::Duration;
4
5use crate::config::build_vv_llm_from_local_settings;
6use crate::context::RunContext;
7use crate::llm::LlmClient;
8use crate::model::ModelRef;
9use crate::runtime::backends::InlineBackend;
10use crate::runtime::context::ExecutionContext;
11use crate::runtime::engine::{AgentRuntime, RuntimeEventHandler, RuntimeRunControls};
12use crate::runtime::state::{Checkpoint, LeaseOperationClock, StateStore};
13use crate::runtime::tool_planner::project_tool_policy;
14use crate::types::{AgentResult, AgentStatus, Metadata};
15use crate::workspace::LocalWorkspaceBackend;
16
17use super::capabilities::{DistributedCapabilityRegistry, ResolvedDistributedCapabilities};
18use super::contract::{now_unix_ms, DistributedRunEnvelope};
19use super::dispatch::CycleDispatchResult;
20use super::worker_v2::{
21 run_distributed_cycle_v2, DistributedDeliveryMetadata, DistributedV2CycleExecutor,
22};
23
24pub(super) struct LeaseHeartbeatStopGuard {
25 stopped: Arc<(Mutex<bool>, Condvar)>,
26}
27
28impl LeaseHeartbeatStopGuard {
29 pub(super) fn new(stopped: Arc<(Mutex<bool>, Condvar)>) -> Self {
30 Self { stopped }
31 }
32}
33
34impl Drop for LeaseHeartbeatStopGuard {
35 fn drop(&mut self) {
36 let (lock, changed) = &*self.stopped;
37 *lock
38 .lock()
39 .unwrap_or_else(std::sync::PoisonError::into_inner) = true;
40 changed.notify_all();
41 }
42}
43
44#[derive(Clone)]
45pub(super) struct LeaseHeartbeatStatus {
46 state: Arc<Mutex<LeaseHeartbeatState>>,
47 #[cfg(test)]
48 failure_recorded: Arc<Condvar>,
49}
50
51#[derive(Clone, Copy, Debug, PartialEq, Eq)]
52pub(super) enum LeaseCommitPhase {
53 NotStarted,
54 InProgress,
55 Succeeded,
56}
57
58pub(super) struct LeaseHeartbeatFailure {
59 pub(super) renewal: LeaseRenewalFailure,
60 pub(super) renewal_started_during_commit: bool,
61}
62
63struct LeaseHeartbeatState {
64 failure: Option<LeaseHeartbeatFailure>,
65 commit_phase: LeaseCommitPhase,
66}
67
68#[derive(Clone, Copy, Debug, PartialEq, Eq)]
69pub(super) enum LeaseRenewalFailureKind {
70 ActiveClaimLost,
71 ClaimLeaseExpired,
72 Coordination,
73}
74
75pub(super) struct LeaseRenewalFailure {
76 pub(super) kind: LeaseRenewalFailureKind,
77 pub(super) message: String,
78}
79
80impl LeaseRenewalFailure {
81 pub(super) fn active_claim_lost() -> Self {
82 Self {
83 kind: LeaseRenewalFailureKind::ActiveClaimLost,
84 message: "claim is no longer active".to_string(),
85 }
86 }
87
88 pub(super) fn claim_lease_expired() -> Self {
89 Self {
90 kind: LeaseRenewalFailureKind::ClaimLeaseExpired,
91 message: "claim lease expired".to_string(),
92 }
93 }
94
95 pub(super) fn coordination(message: String) -> Self {
96 Self {
97 kind: LeaseRenewalFailureKind::Coordination,
98 message,
99 }
100 }
101}
102
103pub(super) struct LeaseRenewal {
104 pub(super) lease_expires_at_ms: u64,
105 pub(super) effective_lease_ms: u64,
106}
107
108struct LeaseRenewalRequest {
109 now_ms: u64,
110 lease_expires_at_ms: u64,
111 clock: LeaseOperationClock,
112}
113
114impl LeaseHeartbeatStatus {
115 pub(super) fn new() -> Self {
116 Self {
117 state: Arc::new(Mutex::new(LeaseHeartbeatState {
118 failure: None,
119 commit_phase: LeaseCommitPhase::NotStarted,
120 })),
121 #[cfg(test)]
122 failure_recorded: Arc::new(Condvar::new()),
123 }
124 }
125
126 pub(super) fn commit_phase(&self) -> LeaseCommitPhase {
127 self.state
128 .lock()
129 .unwrap_or_else(std::sync::PoisonError::into_inner)
130 .commit_phase
131 }
132
133 pub(super) fn record(&self, renewal: LeaseRenewalFailure, phase_at_start: LeaseCommitPhase) {
134 let mut state = self
135 .state
136 .lock()
137 .unwrap_or_else(std::sync::PoisonError::into_inner);
138 if state.failure.is_none() {
139 state.failure = Some(LeaseHeartbeatFailure {
140 renewal,
141 renewal_started_during_commit: phase_at_start != LeaseCommitPhase::NotStarted,
142 });
143 #[cfg(test)]
144 self.failure_recorded.notify_all();
145 }
146 }
147
148 #[cfg(test)]
149 fn wait_for_failure(&self, timeout: Duration) -> bool {
150 let state = self
151 .state
152 .lock()
153 .unwrap_or_else(std::sync::PoisonError::into_inner);
154 let (state, _) = self
155 .failure_recorded
156 .wait_timeout_while(state, timeout, |state| state.failure.is_none())
157 .unwrap_or_else(std::sync::PoisonError::into_inner);
158 state.failure.is_some()
159 }
160
161 pub(super) fn begin_commit(&self) -> Result<(), String> {
162 let mut state = self
163 .state
164 .lock()
165 .unwrap_or_else(std::sync::PoisonError::into_inner);
166 if let Some(failure) = &state.failure {
167 return Err(format!(
168 "checkpoint lease heartbeat failed: {}",
169 failure.renewal.message
170 ));
171 }
172 state.commit_phase = LeaseCommitPhase::InProgress;
173 Ok(())
174 }
175
176 pub(super) fn mark_commit_succeeded(&self) -> Result<(), String> {
177 let mut state = self
178 .state
179 .lock()
180 .unwrap_or_else(std::sync::PoisonError::into_inner);
181 if state.commit_phase != LeaseCommitPhase::InProgress {
182 return Err("checkpoint commit phase has not started".to_string());
183 }
184 state.commit_phase = LeaseCommitPhase::Succeeded;
185 Ok(())
186 }
187
188 pub(super) fn take(&self) -> (Option<LeaseHeartbeatFailure>, LeaseCommitPhase) {
189 let mut state = self
190 .state
191 .lock()
192 .unwrap_or_else(std::sync::PoisonError::into_inner);
193 (state.failure.take(), state.commit_phase)
194 }
195}
196
197pub(super) struct LeaseOperationResult<T> {
198 pub(super) value: T,
199 pub(super) claim_committed: bool,
200}
201
202impl<T> LeaseOperationResult<T> {
203 pub(super) fn new(value: T, claim_committed: bool) -> Self {
204 Self {
205 value,
206 claim_committed,
207 }
208 }
209
210 #[cfg(test)]
211 fn uncommitted(value: T) -> Self {
212 Self::new(value, false)
213 }
214}
215
216#[derive(Clone)]
217pub struct DistributedCycleWorker {
218 pub(super) capabilities: DistributedCapabilityRegistry,
219 pub(super) checkpoint_executor: Option<Arc<dyn DistributedV2CycleExecutor>>,
220}
221
222impl Default for DistributedCycleWorker {
223 fn default() -> Self {
224 Self::new(DistributedCapabilityRegistry::new())
225 }
226}
227
228impl DistributedCycleWorker {
229 pub fn new(capabilities: DistributedCapabilityRegistry) -> Self {
230 Self {
231 capabilities,
232 checkpoint_executor: None,
233 }
234 }
235
236 pub fn with_checkpoint_executor(
237 mut self,
238 executor: Arc<dyn DistributedV2CycleExecutor>,
239 ) -> Self {
240 self.checkpoint_executor = Some(executor);
241 self
242 }
243
244 pub fn run_cycle(
245 &self,
246 envelope: DistributedRunEnvelope,
247 ) -> Result<CycleDispatchResult, String> {
248 self.run_cycle_with_delivery(envelope, DistributedDeliveryMetadata::default())
249 }
250
251 pub fn run_cycle_with_delivery(
252 &self,
253 envelope: DistributedRunEnvelope,
254 delivery: DistributedDeliveryMetadata,
255 ) -> Result<CycleDispatchResult, String> {
256 envelope.validate()?;
257 envelope.ensure_not_expired()?;
258 if envelope.is_checkpoint_v2() {
259 return run_distributed_cycle_v2(self, envelope, delivery);
260 }
261 self.run_cycle_v1(envelope)
262 }
263
264 fn run_cycle_v1(
265 &self,
266 envelope: DistributedRunEnvelope,
267 ) -> Result<CycleDispatchResult, String> {
268 let state_store = envelope
269 .recipe
270 .build_state_store()
271 .map_err(|error| error.to_string())?;
272 if let Some(checkpoint) = state_store
273 .load_checkpoint(&envelope.task.task_id)
274 .map_err(|error| error.to_string())?
275 {
276 if let Some(result) = checkpoint.terminal_result {
277 return Ok(CycleDispatchResult::finished_at_revision(
278 result,
279 Some(checkpoint.revision),
280 ));
281 }
282 }
283
284 let resolved = self
286 .capabilities
287 .resolve(&envelope.recipe.capabilities)
288 .map_err(|error| error.to_string())?;
289 let runtime = build_runtime(&envelope, &resolved)?;
290 let heartbeat_state_store = envelope
291 .recipe
292 .build_state_store()
293 .map_err(|error| error.to_string())?;
294 let now_ms = now_unix_ms()?;
295 envelope.ensure_not_expired_at(now_ms)?;
296 let lease_expires_at_ms = lease_expiry_at(
297 now_ms,
298 envelope.lease_duration_ms,
299 envelope.deadline_unix_ms,
300 )?;
301 let claim_token = uuid::Uuid::new_v4().simple().to_string();
302 let Some(mut checkpoint) = state_store
303 .claim_checkpoint(
304 &envelope.task.task_id,
305 envelope.cycle_index,
306 &claim_token,
307 lease_expires_at_ms,
308 now_ms,
309 )
310 .map_err(|error| format!("retryable distributed delivery conflict: {error}"))?
311 else {
312 return Ok(CycleDispatchResult::finished(failed_result(
313 format!("No checkpoint found for task {}", envelope.task.task_id),
314 Vec::new(),
315 Vec::new(),
316 Metadata::new(),
317 )));
318 };
319
320 let previous_cycle_count = checkpoint.cycles.len();
321 let controls = worker_controls(&envelope, &resolved, &checkpoint, state_store.clone());
322 let mut worker_task = envelope.task.clone();
323 project_tool_policy(&mut worker_task, &resolved.tool_policy);
324 let cycle_result = run_with_lease_heartbeat(
325 heartbeat_state_store,
326 &envelope,
327 &claim_token,
328 checkpoint.revision,
329 |heartbeat_status| {
330 let cycle_result = (|| -> Result<CycleDispatchResult, String> {
331 let runtime_result = runtime.run_with_controls(worker_task, controls);
332 let result = runtime_result.unwrap_or_else(|error| {
333 failed_result(
334 error.to_string(),
335 checkpoint.messages.clone(),
336 checkpoint.cycles.clone(),
337 checkpoint.shared_state.clone(),
338 )
339 });
340
341 checkpoint.cycle_index = envelope.cycle_index;
342 checkpoint.messages = result.messages.clone();
343 checkpoint.cycles = result.cycles.clone();
344 checkpoint.shared_state = result.shared_state.clone();
345 checkpoint.budget_usage = result.budget_usage.clone();
346 let expected_revision = checkpoint.revision;
347 heartbeat_status.begin_commit()?;
348 if result.status == AgentStatus::MaxCycles
349 && result.cycles.len() > previous_cycle_count
350 {
351 checkpoint.status = AgentStatus::Running;
352 checkpoint.terminal_result = None;
353 if !state_store
354 .commit_checkpoint(checkpoint, &claim_token, expected_revision)
355 .map_err(|error| error.to_string())?
356 {
357 return Err(format!(
358 "checkpoint changed while cycle {} was running for task {}",
359 envelope.cycle_index, envelope.task.task_id
360 ));
361 }
362 heartbeat_status.mark_commit_succeeded()?;
363 return Ok(CycleDispatchResult::unfinished());
364 }
365
366 checkpoint.status = result.status;
367 checkpoint.terminal_result = Some(result.clone());
368 if !state_store
369 .commit_checkpoint(checkpoint, &claim_token, expected_revision)
370 .map_err(|error| error.to_string())?
371 {
372 return Err(format!(
373 "checkpoint changed while terminal cycle {} was running for task {}",
374 envelope.cycle_index, envelope.task.task_id
375 ));
376 }
377 heartbeat_status.mark_commit_succeeded()?;
378 Ok(CycleDispatchResult::finished_at_revision(
379 result,
380 Some(expected_revision + 1),
381 ))
382 })();
383 let claim_committed = cycle_result.is_ok();
384 LeaseOperationResult::new(cycle_result, claim_committed)
385 },
386 )?;
387 cycle_result
388 }
389}
390
391fn run_with_lease_heartbeat<T>(
392 state_store: Arc<dyn StateStore>,
393 envelope: &DistributedRunEnvelope,
394 claim_token: &str,
395 expected_revision: u64,
396 operation: impl FnOnce(&LeaseHeartbeatStatus) -> LeaseOperationResult<T>,
397) -> Result<T, String> {
398 run_with_checkpoint_lease(
399 state_store.as_ref(),
400 &envelope.task.task_id,
401 envelope.cycle_index,
402 envelope.lease_duration_ms,
403 envelope.deadline_unix_ms,
404 claim_token,
405 expected_revision,
406 operation,
407 )
408}
409
410#[allow(clippy::too_many_arguments)]
411pub(super) fn run_with_checkpoint_lease<T>(
412 state_store: &dyn StateStore,
413 task_id: &str,
414 cycle_index: u32,
415 lease_duration_ms: u64,
416 deadline_unix_ms: Option<u64>,
417 claim_token: &str,
418 expected_revision: u64,
419 operation: impl FnOnce(&LeaseHeartbeatStatus) -> LeaseOperationResult<T>,
420) -> Result<T, String> {
421 let stopped = Arc::new((Mutex::new(false), Condvar::new()));
422 let heartbeat_status = LeaseHeartbeatStatus::new();
423 let task_id = task_id.to_string();
424 let claim_token = claim_token.to_string();
425
426 let known_lease_expires_at_ms = load_claim_lease_expiry(
427 state_store,
428 &task_id,
429 &claim_token,
430 expected_revision,
431 cycle_index,
432 )
433 .map_err(|failure| format!("checkpoint lease heartbeat failed: {}", failure.message))?;
434 let initial_request = prepare_lease_renewal(&task_id, lease_duration_ms, deadline_unix_ms)
435 .map_err(|failure| format!("checkpoint lease heartbeat failed: {}", failure.message))?;
436 let initial_renewal = renew_checkpoint_lease(
437 state_store,
438 &task_id,
439 &claim_token,
440 expected_revision,
441 initial_request,
442 known_lease_expires_at_ms,
443 )
444 .map_err(|failure| format!("checkpoint lease heartbeat failed: {}", failure.message))?;
445
446 let result = std::thread::scope(|scope| {
447 let stopped_for_thread = stopped.clone();
448 let status_for_thread = heartbeat_status.clone();
449 let heartbeat = scope.spawn(move || {
450 let mut known_lease_expires_at_ms = initial_renewal.lease_expires_at_ms;
451 let mut interval = lease_heartbeat_interval(initial_renewal.effective_lease_ms);
452 loop {
453 let (lock, changed) = &*stopped_for_thread;
454 let guard = lock
455 .lock()
456 .unwrap_or_else(std::sync::PoisonError::into_inner);
457 let (guard, _) = changed
458 .wait_timeout_while(guard, interval, |stopped| !*stopped)
459 .unwrap_or_else(std::sync::PoisonError::into_inner);
460 if *guard {
461 break;
462 }
463 drop(guard);
464
465 let request =
466 match prepare_lease_renewal(&task_id, lease_duration_ms, deadline_unix_ms) {
467 Ok(request) => request,
468 Err(failure) => {
469 status_for_thread.record(failure, LeaseCommitPhase::NotStarted);
470 break;
471 }
472 };
473 let phase_at_start = status_for_thread.commit_phase();
474 let outcome = renew_checkpoint_lease(
475 state_store,
476 &task_id,
477 &claim_token,
478 expected_revision,
479 request,
480 known_lease_expires_at_ms,
481 );
482 match outcome {
483 Ok(renewal) => {
484 known_lease_expires_at_ms = renewal.lease_expires_at_ms;
485 interval = lease_heartbeat_interval(renewal.effective_lease_ms);
486 }
487 Err(failure) => {
488 status_for_thread.record(failure, phase_at_start);
489 break;
490 }
491 }
492 }
493 });
494
495 let stop_guard = LeaseHeartbeatStopGuard::new(stopped.clone());
496 let result = operation(&heartbeat_status);
497 drop(stop_guard);
498 heartbeat
499 .join()
500 .map_err(|_| "checkpoint lease heartbeat panicked".to_string())?;
501 Ok::<_, String>(result)
502 })?;
503
504 let (failure, commit_phase) = heartbeat_status.take();
505 if let Some(failure) = failure {
506 let commit_consumed_claim = result.claim_committed
507 && commit_phase == LeaseCommitPhase::Succeeded
508 && failure.renewal_started_during_commit
509 && failure.renewal.kind == LeaseRenewalFailureKind::ActiveClaimLost;
510 if !commit_consumed_claim {
511 return Err(format!(
512 "checkpoint lease heartbeat failed: {}",
513 failure.renewal.message
514 ));
515 }
516 }
517 Ok(result.value)
518}
519
520fn load_claim_lease_expiry(
521 state_store: &dyn StateStore,
522 task_id: &str,
523 claim_token: &str,
524 expected_revision: u64,
525 expected_cycle: u32,
526) -> Result<u64, LeaseRenewalFailure> {
527 let checkpoint = state_store
528 .load_checkpoint(task_id)
529 .map_err(|error| LeaseRenewalFailure::coordination(error.to_string()))?;
530 checkpoint
531 .filter(|checkpoint| {
532 checkpoint.revision == expected_revision
533 && checkpoint.claim_token.as_deref() == Some(claim_token)
534 && checkpoint.claimed_cycle == Some(expected_cycle)
535 })
536 .and_then(|checkpoint| checkpoint.lease_expires_at_ms)
537 .ok_or_else(LeaseRenewalFailure::active_claim_lost)
538}
539
540fn prepare_lease_renewal(
541 task_id: &str,
542 lease_duration_ms: u64,
543 deadline_unix_ms: Option<u64>,
544) -> Result<LeaseRenewalRequest, LeaseRenewalFailure> {
545 let now_ms = now_unix_ms().map_err(LeaseRenewalFailure::coordination)?;
546 let clock = LeaseOperationClock::new(now_ms);
547 if deadline_unix_ms.is_some_and(|deadline| deadline <= now_ms) {
548 return Err(LeaseRenewalFailure::coordination(format!(
549 "distributed job deadline expired while renewing {task_id}"
550 )));
551 }
552 let lease_expires_at_ms = lease_expiry_at(now_ms, lease_duration_ms, deadline_unix_ms)
553 .map_err(LeaseRenewalFailure::coordination)?;
554 Ok(LeaseRenewalRequest {
555 now_ms,
556 lease_expires_at_ms,
557 clock,
558 })
559}
560
561fn renew_checkpoint_lease(
562 state_store: &dyn StateStore,
563 task_id: &str,
564 claim_token: &str,
565 expected_revision: u64,
566 request: LeaseRenewalRequest,
567 known_lease_expires_at_ms: u64,
568) -> Result<LeaseRenewal, LeaseRenewalFailure> {
569 let renewed = state_store
570 .renew_checkpoint_claim(
571 task_id,
572 claim_token,
573 expected_revision,
574 request.lease_expires_at_ms,
575 request.now_ms,
576 )
577 .map_err(|error| {
578 let message = error.to_string();
579 if message == "claim lease expired" {
580 LeaseRenewalFailure::claim_lease_expired()
581 } else {
582 LeaseRenewalFailure::coordination(message)
583 }
584 })?;
585 let observed_at_ms = request
586 .clock
587 .now_ms()
588 .max(now_unix_ms().map_err(LeaseRenewalFailure::coordination)?);
589 if !renewed {
590 return Err(
591 if observed_at_ms >= known_lease_expires_at_ms
592 || observed_at_ms >= request.lease_expires_at_ms
593 {
594 LeaseRenewalFailure::claim_lease_expired()
595 } else {
596 LeaseRenewalFailure::active_claim_lost()
597 },
598 );
599 }
600 if observed_at_ms >= known_lease_expires_at_ms || observed_at_ms >= request.lease_expires_at_ms
601 {
602 return Err(LeaseRenewalFailure::claim_lease_expired());
603 }
604 Ok(LeaseRenewal {
605 lease_expires_at_ms: request.lease_expires_at_ms,
606 effective_lease_ms: request.lease_expires_at_ms - observed_at_ms,
607 })
608}
609
610fn lease_heartbeat_interval(lease_duration_ms: u64) -> Duration {
611 let interval_micros = lease_duration_ms
612 .saturating_mul(1_000)
613 .saturating_div(3)
614 .clamp(1, 30_000_000);
615 Duration::from_micros(interval_micros)
616}
617
618pub(super) fn lease_expiry_at(
619 now_ms: u64,
620 lease_duration_ms: u64,
621 deadline_unix_ms: Option<u64>,
622) -> Result<u64, String> {
623 let effective_duration = deadline_unix_ms
624 .map(|deadline| lease_duration_ms.min(deadline.saturating_sub(now_ms)))
625 .unwrap_or(lease_duration_ms);
626 now_ms
627 .checked_add(effective_duration)
628 .ok_or_else(|| "checkpoint lease overflow".to_string())
629}
630
631pub(super) fn build_runtime(
632 envelope: &DistributedRunEnvelope,
633 resolved: &ResolvedDistributedCapabilities,
634) -> Result<AgentRuntime<Arc<dyn LlmClient>>, String> {
635 let llm_client = match resolved.llm_client.clone() {
636 Some(client) => client,
637 None => Arc::new(
638 build_vv_llm_from_local_settings(
639 &envelope.recipe.settings_file,
640 &envelope.recipe.backend,
641 &envelope.recipe.model,
642 envelope.recipe.timeout_seconds,
643 )
644 .map_err(|error| error.to_string())?
645 .0,
646 ) as Arc<dyn LlmClient>,
647 };
648 let workspace = PathBuf::from(&envelope.recipe.workspace);
649 let workspace_backend = resolved
650 .workspace_backend
651 .clone()
652 .unwrap_or_else(|| Arc::new(LocalWorkspaceBackend::new(workspace.clone())));
653 let mut runtime = AgentRuntime::new(llm_client)
654 .with_tool_registry(resolved.tool_registry.clone())
655 .with_execution_backend(InlineBackend)
656 .with_default_workspace(workspace)
657 .with_workspace_backend(workspace_backend)
658 .with_settings_file(&envelope.recipe.settings_file)
659 .with_default_backend(&envelope.recipe.backend)
660 .with_hooks(resolved.hooks.clone());
661 if let Some(log_preview_chars) = envelope.recipe.log_preview_chars {
662 runtime = runtime.with_log_preview_chars(log_preview_chars);
663 }
664 runtime.set_tool_policy(resolved.tool_policy.clone());
665 Ok(runtime)
666}
667
668fn worker_controls(
669 envelope: &DistributedRunEnvelope,
670 resolved: &ResolvedDistributedCapabilities,
671 checkpoint: &Checkpoint,
672 state_store: Arc<dyn StateStore>,
673) -> RuntimeRunControls {
674 let mut metadata = envelope.task.metadata.clone();
675 metadata.insert(
676 "_vv_agent_run_id".to_string(),
677 serde_json::Value::String(envelope.run_id.clone()),
678 );
679 let mut execution_context = ExecutionContext {
680 cancellation_token: resolved.cancellation.clone(),
681 state_store: Some(state_store),
682 approval_provider: resolved.approval_provider.clone(),
683 approval_broker: resolved.approval_broker.clone(),
684 approval_timeout: resolved
685 .approval_timeout_seconds
686 .map(Duration::from_secs_f64),
687 memory_providers: resolved.memory_providers.clone(),
688 app_state: resolved.app_state.clone(),
689 metadata,
690 ..ExecutionContext::default()
691 };
692 if execution_context.approval_provider.is_some() && execution_context.approval_broker.is_none()
693 {
694 execution_context.approval_broker = Some(Default::default());
695 }
696 RuntimeRunControls {
697 log_handler: combined_event_handler(resolved),
698 cancellation_token: resolved.cancellation.clone(),
699 execution_context: Some(execution_context),
700 workspace: Some(PathBuf::from(&envelope.recipe.workspace)),
701 workspace_backend: resolved.workspace_backend.clone(),
702 run_context: Some(RunContext {
703 run_id: envelope.run_id.clone(),
704 model: Some(ModelRef::backend(
705 envelope.recipe.backend.clone(),
706 envelope.recipe.model.clone(),
707 )),
708 workspace: Some(PathBuf::from(&envelope.recipe.workspace)),
709 app_state: resolved.app_state.clone(),
710 ..RunContext::default()
711 }),
712 sub_task_manager: resolved.sub_task_manager.clone(),
713 budget_limits: envelope.budget_limits.clone(),
714 host_cost_meter: resolved.host_cost_meter.clone(),
715 initial_messages: Some(checkpoint.messages.clone()),
716 initial_shared_state: Some(checkpoint.shared_state.clone()),
717 initial_cycles: Some(checkpoint.cycles.clone()),
718 cycle_index_start: Some(envelope.cycle_index),
719 cycle_count: Some(1),
720 initial_budget_usage: checkpoint.budget_usage.clone(),
721 defer_terminal_on_max_cycles: true,
722 ..RuntimeRunControls::default()
723 }
724}
725
726pub(super) fn combined_event_handler(
727 resolved: &ResolvedDistributedCapabilities,
728) -> Option<RuntimeEventHandler> {
729 let mut handlers = resolved.observers.clone();
730 if let Some(event_sink) = &resolved.event_sink {
731 handlers.push(event_sink.clone());
732 }
733 if handlers.is_empty() {
734 return None;
735 }
736 Some(Arc::new(move |event, payload| {
737 for handler in &handlers {
738 handler(event, payload);
739 }
740 }))
741}
742
743fn failed_result(
744 error: String,
745 messages: Vec<crate::types::Message>,
746 cycles: Vec<crate::types::CycleRecord>,
747 shared_state: Metadata,
748) -> AgentResult {
749 let token_usage = crate::runtime::summarize_task_token_usage(&cycles);
750 let partial_output = crate::types::last_assistant_output(&cycles);
751 AgentResult {
752 status: AgentStatus::Failed,
753 messages,
754 cycles,
755 completion_reason: Some(crate::types::CompletionReason::Failed),
756 completion_tool_name: None,
757 partial_output,
758 budget_usage: None,
759 budget_exhaustion: None,
760 checkpoint_key: None,
761 resume_observation: None,
762 final_answer: None,
763 wait_reason: None,
764 error: Some(error),
765 shared_state,
766 token_usage,
767 }
768}
769
770#[cfg(test)]
771mod tests;