1use std::collections::BTreeSet;
2
3use serde::{Deserialize, Serialize};
4use serde_json::Value;
5
6use crate::budget::{RunBudgetLimits, MAX_WIRE_INTEGER};
7use crate::checkpoint::{
8 validate_checkpoint_key, validate_extension_namespace, validate_sha256, AmbiguousModelPolicy,
9 AmbiguousToolPolicy, ClaimMode, ResumePolicy, DEFAULT_MAX_EXTENSION_STATE_BYTES,
10 RUN_DEFINITION_SCHEMA,
11};
12use crate::tools::{ToolPolicy, ToolSideEffect};
13use crate::types::AgentTask;
14
15use super::super::RuntimeRecipe;
16pub(crate) use super::contract_helpers::now_unix_ms;
17use super::contract_helpers::{
18 normalize_integral_float, require_non_empty, utf16_cmp, validate_current_discriminator_fields,
19 validate_json_pointer, validate_sorted_unique,
20};
21
22pub const DISTRIBUTED_RUN_SCHEMA_VERSION: &str = "vv-agent.distributed-run.v2";
23pub const DEFAULT_TOOLSET_ID: &str = "vv-agent.builtin-tools";
24pub const DEFAULT_TOOLSET_VERSION: &str = "1";
25pub const DEFAULT_TOOLSET_SCHEMA_DIGEST: &str =
26 "24d8f7bde18b11374820f742cfa244c83666626a315e09d4b6e1b69e899a70aa";
27pub const DEFAULT_CYCLE_NAME: &str = "vv_agent.distributed.run_single_cycle";
28pub const DEFAULT_LEASE_DURATION_MS: u64 = 5 * 60 * 1000;
29
30#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
31pub struct CapabilityRef {
32 pub id: String,
33 pub version: String,
34}
35
36impl CapabilityRef {
37 pub fn new(id: impl Into<String>, version: impl Into<String>) -> Result<Self, String> {
38 let reference = Self {
39 id: id.into(),
40 version: version.into(),
41 };
42 reference.validate("capability_ref")?;
43 Ok(reference)
44 }
45
46 pub fn validate(&self, field_name: &str) -> Result<(), String> {
47 require_non_empty(&self.id, &format!("{field_name}.id"))?;
48 require_non_empty(&self.version, &format!("{field_name}.version"))
49 }
50
51 pub fn to_dict(&self) -> Value {
52 serde_json::json!({"id": self.id, "version": self.version})
53 }
54
55 pub fn from_dict(payload: &Value, field_name: &str) -> Result<Self, String> {
56 let reference: Self = serde_json::from_value(payload.clone())
57 .map_err(|_| format!("{field_name} must be an object"))?;
58 reference.validate(field_name)?;
59 Ok(reference)
60 }
61}
62
63#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
64pub struct ToolsetRef {
65 pub id: String,
66 pub version: String,
67 pub schema_digest: String,
68}
69
70impl Default for ToolsetRef {
71 fn default() -> Self {
72 Self {
73 id: DEFAULT_TOOLSET_ID.to_string(),
74 version: DEFAULT_TOOLSET_VERSION.to_string(),
75 schema_digest: DEFAULT_TOOLSET_SCHEMA_DIGEST.to_string(),
76 }
77 }
78}
79
80impl ToolsetRef {
81 pub fn validate(&self) -> Result<(), String> {
82 CapabilityRef {
83 id: self.id.clone(),
84 version: self.version.clone(),
85 }
86 .validate("toolset_ref")?;
87 if self.schema_digest.len() != 64
88 || self
89 .schema_digest
90 .bytes()
91 .any(|byte| !byte.is_ascii_hexdigit() || byte.is_ascii_uppercase())
92 {
93 return Err(
94 "toolset_ref.schema_digest must be a lowercase SHA-256 hex digest".to_string(),
95 );
96 }
97 Ok(())
98 }
99
100 pub fn capability_ref(&self) -> CapabilityRef {
101 CapabilityRef {
102 id: self.id.clone(),
103 version: self.version.clone(),
104 }
105 }
106
107 pub fn to_dict(&self) -> Value {
108 serde_json::json!({
109 "id": self.id,
110 "version": self.version,
111 "schema_digest": self.schema_digest,
112 })
113 }
114
115 pub fn from_dict(payload: &Value) -> Result<Self, String> {
116 let reference: Self = serde_json::from_value(payload.clone())
117 .map_err(|_| "toolset_ref must be an object".to_string())?;
118 reference.validate()?;
119 Ok(reference)
120 }
121}
122
123#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
124pub struct DistributedToolPolicy {
125 pub allowed_tools: Option<Vec<String>>,
126 #[serde(default)]
127 pub disallowed_tools: Vec<String>,
128 #[serde(default = "default_approval_policy")]
129 pub approval: String,
130 #[serde(default)]
131 pub predicate_ref: Option<CapabilityRef>,
132 #[serde(default)]
133 pub denied_side_effects: Vec<ToolSideEffect>,
134 #[serde(default)]
135 pub denied_capability_tags: Vec<String>,
136 #[serde(default)]
137 pub deny_terminal_tools: bool,
138 #[serde(default)]
139 pub denied_cost_dimensions: Vec<String>,
140}
141
142impl Default for DistributedToolPolicy {
143 fn default() -> Self {
144 Self {
145 allowed_tools: None,
146 disallowed_tools: Vec::new(),
147 approval: default_approval_policy(),
148 predicate_ref: None,
149 denied_side_effects: Vec::new(),
150 denied_capability_tags: Vec::new(),
151 deny_terminal_tools: false,
152 denied_cost_dimensions: Vec::new(),
153 }
154 }
155}
156
157impl DistributedToolPolicy {
158 pub(crate) fn set_metadata_denials(&mut self, policy: &ToolPolicy) {
159 self.denied_side_effects = policy.denied_side_effects.clone();
160 self.denied_capability_tags = policy.denied_capability_tags.clone();
161 self.deny_terminal_tools = policy.deny_terminal_tools;
162 self.denied_cost_dimensions = policy.denied_cost_dimensions.clone();
163 }
164
165 fn metadata_denials(&self) -> Result<ToolPolicy, String> {
166 ToolPolicy {
167 denied_side_effects: self.denied_side_effects.clone(),
168 denied_capability_tags: self.denied_capability_tags.clone(),
169 deny_terminal_tools: self.deny_terminal_tools,
170 denied_cost_dimensions: self.denied_cost_dimensions.clone(),
171 ..ToolPolicy::default()
172 }
173 .normalized()
174 .map_err(|error| error.to_string())
175 }
176
177 pub fn validate(&self) -> Result<(), String> {
178 if !matches!(
179 self.approval.as_str(),
180 "default" | "always" | "never" | "on_request"
181 ) {
182 return Err("tool_policy.approval is unsupported".to_string());
183 }
184 for (field_name, values) in [
185 ("tool_policy.allowed_tools", self.allowed_tools.as_deref()),
186 (
187 "tool_policy.disallowed_tools",
188 Some(self.disallowed_tools.as_slice()),
189 ),
190 ] {
191 if values.is_some_and(|values| values.iter().any(|value| value.trim().is_empty())) {
192 return Err(format!("{field_name} must contain non-empty strings"));
193 }
194 }
195 if let Some(reference) = &self.predicate_ref {
196 reference.validate("tool_policy.predicate_ref")?;
197 }
198 self.metadata_denials()?;
199 Ok(())
200 }
201
202 pub fn to_dict(&self) -> Value {
203 serde_json::json!({
204 "allowed_tools": self.allowed_tools,
205 "disallowed_tools": self.disallowed_tools,
206 "approval": self.approval,
207 "predicate_ref": self.predicate_ref,
208 "denied_side_effects": self.denied_side_effects,
209 "denied_capability_tags": self.denied_capability_tags,
210 "deny_terminal_tools": self.deny_terminal_tools,
211 "denied_cost_dimensions": self.denied_cost_dimensions,
212 })
213 }
214
215 pub fn from_dict(payload: &Value) -> Result<Self, String> {
216 let policy: Self = serde_json::from_value(payload.clone())
217 .map_err(|_| "tool_policy must be an object".to_string())?;
218 policy.validate()?;
219 Ok(policy)
220 }
221}
222
223fn default_approval_policy() -> String {
224 "default".to_string()
225}
226
227#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
228pub struct DistributedCheckpointExtensionRef {
229 pub namespace: String,
230 pub reference: CapabilityRef,
231 #[serde(default)]
232 pub required: bool,
233}
234
235impl DistributedCheckpointExtensionRef {
236 pub fn validate(&self) -> Result<(), String> {
237 validate_extension_namespace(&self.namespace).map_err(|error| error.to_string())?;
238 self.reference
239 .validate("checkpoint_extension_ref.reference")
240 }
241}
242
243#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
244pub struct DistributedCheckpointConfig {
245 pub key: String,
246 #[serde(default)]
247 pub resume_policy: ResumePolicy,
248 #[serde(default)]
249 pub ambiguous_model_policy: AmbiguousModelPolicy,
250 #[serde(default)]
251 pub ambiguous_tool_policy: AmbiguousToolPolicy,
252 #[serde(default)]
253 pub required_extension_namespaces: Vec<String>,
254 #[serde(default = "default_max_extension_state_bytes")]
255 pub max_extension_state_bytes: u64,
256 #[serde(default)]
257 pub credential_slots: Vec<String>,
258}
259
260impl DistributedCheckpointConfig {
261 pub fn validate(&self) -> Result<(), String> {
262 validate_checkpoint_key(&self.key).map_err(|error| error.to_string())?;
263 if self.max_extension_state_bytes > MAX_WIRE_INTEGER {
264 return Err(format!(
265 "max_extension_state_bytes must be between 0 and {MAX_WIRE_INTEGER}"
266 ));
267 }
268 validate_sorted_unique(
269 &self.required_extension_namespaces,
270 "required_extension_namespaces",
271 |namespace| validate_extension_namespace(namespace).map_err(|error| error.to_string()),
272 )?;
273 for pointer in &self.credential_slots {
274 validate_json_pointer(pointer)?;
275 }
276 if self
277 .credential_slots
278 .windows(2)
279 .any(|window| utf16_cmp(&window[0], &window[1]) != std::cmp::Ordering::Less)
280 {
281 return Err("credential_slots must be sorted and unique".to_string());
282 }
283 Ok(())
284 }
285
286 pub fn to_dict(&self) -> Value {
287 serde_json::to_value(self).expect("validated checkpoint config always serializes")
288 }
289}
290
291fn default_max_extension_state_bytes() -> u64 {
292 DEFAULT_MAX_EXTENSION_STATE_BYTES
293}
294
295#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
296pub struct DistributedCapabilities {
297 pub toolset_ref: ToolsetRef,
298 pub tool_policy: DistributedToolPolicy,
299 pub llm_client_ref: Option<CapabilityRef>,
300 pub workspace_backend_ref: Option<CapabilityRef>,
301 pub approval_provider_ref: Option<CapabilityRef>,
302 pub approval_broker_ref: Option<CapabilityRef>,
303 pub approval_timeout_seconds: Option<f64>,
304 pub cancellation_ref: Option<CapabilityRef>,
305 pub event_sink_ref: Option<CapabilityRef>,
306 pub host_cost_meter_ref: Option<CapabilityRef>,
307 pub app_state_ref: Option<CapabilityRef>,
308 pub sub_task_manager_ref: Option<CapabilityRef>,
309 #[serde(default)]
310 pub memory_provider_refs: Vec<CapabilityRef>,
311 #[serde(default)]
312 pub hook_refs: Vec<CapabilityRef>,
313 #[serde(default, skip_serializing_if = "Vec::is_empty")]
314 pub after_cycle_hook_refs: Vec<CapabilityRef>,
315 #[serde(default)]
316 pub observer_refs: Vec<CapabilityRef>,
317 #[serde(default, skip_serializing_if = "Option::is_none")]
318 pub checkpoint_store_ref: Option<CapabilityRef>,
319 #[serde(default, skip_serializing_if = "Option::is_none")]
320 pub checkpoint_event_store_ref: Option<CapabilityRef>,
321 #[serde(default, skip_serializing_if = "Vec::is_empty")]
322 pub checkpoint_extension_refs: Vec<DistributedCheckpointExtensionRef>,
323 #[serde(default, skip_serializing_if = "Option::is_none")]
324 pub reconciliation_provider_ref: Option<CapabilityRef>,
325}
326
327impl DistributedCapabilities {
328 pub fn validate(&self) -> Result<(), String> {
329 self.toolset_ref.validate()?;
330 self.tool_policy.validate()?;
331 if self.approval_provider_ref.is_some() != self.approval_broker_ref.is_some() {
332 return Err(
333 "approval_provider_ref and approval_broker_ref must be declared together"
334 .to_string(),
335 );
336 }
337 if self
338 .approval_timeout_seconds
339 .is_some_and(|value| !value.is_finite() || value <= 0.0)
340 {
341 return Err(
342 "approval_timeout_seconds must be a finite positive number or null".to_string(),
343 );
344 }
345 for (field_name, reference) in [
346 ("llm_client_ref", self.llm_client_ref.as_ref()),
347 ("workspace_backend_ref", self.workspace_backend_ref.as_ref()),
348 ("approval_provider_ref", self.approval_provider_ref.as_ref()),
349 ("approval_broker_ref", self.approval_broker_ref.as_ref()),
350 ("cancellation_ref", self.cancellation_ref.as_ref()),
351 ("event_sink_ref", self.event_sink_ref.as_ref()),
352 ("host_cost_meter_ref", self.host_cost_meter_ref.as_ref()),
353 ("app_state_ref", self.app_state_ref.as_ref()),
354 ("sub_task_manager_ref", self.sub_task_manager_ref.as_ref()),
355 ("checkpoint_store_ref", self.checkpoint_store_ref.as_ref()),
356 (
357 "checkpoint_event_store_ref",
358 self.checkpoint_event_store_ref.as_ref(),
359 ),
360 (
361 "reconciliation_provider_ref",
362 self.reconciliation_provider_ref.as_ref(),
363 ),
364 ] {
365 if let Some(reference) = reference {
366 reference.validate(field_name)?;
367 }
368 }
369 for (field_name, references) in [
370 ("memory_provider_refs", self.memory_provider_refs.as_slice()),
371 ("hook_refs", self.hook_refs.as_slice()),
372 (
373 "after_cycle_hook_refs",
374 self.after_cycle_hook_refs.as_slice(),
375 ),
376 ("observer_refs", self.observer_refs.as_slice()),
377 ] {
378 for (index, reference) in references.iter().enumerate() {
379 reference.validate(&format!("capabilities.{field_name}[{index}]"))?;
380 }
381 }
382 let mut namespaces = BTreeSet::new();
383 for reference in &self.checkpoint_extension_refs {
384 reference.validate()?;
385 if !namespaces.insert(reference.namespace.as_str()) {
386 return Err(format!(
387 "duplicate checkpoint extension namespace {}",
388 reference.namespace
389 ));
390 }
391 }
392 Ok(())
393 }
394
395 pub fn to_dict(&self) -> Value {
396 let mut value = serde_json::json!({
397 "toolset_ref": self.toolset_ref,
398 "tool_policy": self.tool_policy,
399 "llm_client_ref": self.llm_client_ref,
400 "workspace_backend_ref": self.workspace_backend_ref,
401 "approval_provider_ref": self.approval_provider_ref,
402 "approval_broker_ref": self.approval_broker_ref,
403 "approval_timeout_seconds": self.approval_timeout_seconds,
404 "cancellation_ref": self.cancellation_ref,
405 "event_sink_ref": self.event_sink_ref,
406 "host_cost_meter_ref": self.host_cost_meter_ref,
407 "app_state_ref": self.app_state_ref,
408 "sub_task_manager_ref": self.sub_task_manager_ref,
409 "memory_provider_refs": self.memory_provider_refs,
410 "hook_refs": self.hook_refs,
411 "observer_refs": self.observer_refs,
412 });
413 let object = value
414 .as_object_mut()
415 .expect("distributed capabilities are always an object");
416 for (name, reference) in [
417 ("checkpoint_store_ref", self.checkpoint_store_ref.as_ref()),
418 (
419 "checkpoint_event_store_ref",
420 self.checkpoint_event_store_ref.as_ref(),
421 ),
422 (
423 "reconciliation_provider_ref",
424 self.reconciliation_provider_ref.as_ref(),
425 ),
426 ] {
427 if let Some(reference) = reference {
428 object.insert(name.to_string(), reference.to_dict());
429 }
430 }
431 if !self.checkpoint_extension_refs.is_empty() {
432 object.insert(
433 "checkpoint_extension_refs".to_string(),
434 serde_json::to_value(&self.checkpoint_extension_refs)
435 .expect("validated checkpoint extension references always serialize"),
436 );
437 }
438 if !self.after_cycle_hook_refs.is_empty() {
439 object.insert(
440 "after_cycle_hook_refs".to_string(),
441 serde_json::to_value(&self.after_cycle_hook_refs)
442 .expect("validated after-cycle hook references always serialize"),
443 );
444 }
445 value
446 }
447
448 pub fn from_dict(payload: &Value) -> Result<Self, String> {
449 let capabilities: Self = serde_json::from_value(payload.clone())
450 .map_err(|_| "capabilities must be an object".to_string())?;
451 capabilities.validate()?;
452 Ok(capabilities)
453 }
454}
455
456#[derive(Debug, Clone, PartialEq, Deserialize)]
457pub struct DistributedRunEnvelope {
458 pub schema_version: String,
459 pub job_id: String,
460 pub run_id: String,
461 pub task: AgentTask,
462 #[serde(default)]
463 pub budget_limits: Option<RunBudgetLimits>,
464 pub recipe: RuntimeRecipe,
465 pub cycle_name: String,
466 pub cycle_index: u32,
467 pub idempotency_key: String,
468 pub deadline_unix_ms: Option<u64>,
469 pub lease_duration_ms: u64,
470 pub root_run_id: String,
471 pub trace_id: String,
472 pub run_definition_schema: String,
473 pub run_definition_digest: String,
474 pub claim_mode: ClaimMode,
475 pub resume_attempt: u64,
476 pub checkpoint_config: DistributedCheckpointConfig,
477}
478
479impl DistributedRunEnvelope {
480 #[allow(clippy::too_many_arguments)]
481 pub fn for_cycle(
482 task: AgentTask,
483 recipe: RuntimeRecipe,
484 cycle_index: u32,
485 cycle_name: impl Into<String>,
486 run_id: Option<String>,
487 deadline_unix_ms: Option<u64>,
488 lease_duration_ms: u64,
489 budget_limits: Option<RunBudgetLimits>,
490 root_run_id: impl Into<String>,
491 trace_id: impl Into<String>,
492 run_definition_digest: impl Into<String>,
493 claim_mode: ClaimMode,
494 resume_attempt: u64,
495 checkpoint_config: DistributedCheckpointConfig,
496 ) -> Result<Self, String> {
497 let run_id = run_id
498 .filter(|value| !value.trim().is_empty())
499 .or_else(|| {
500 task.metadata
501 .get("_vv_agent_run_id")
502 .and_then(Value::as_str)
503 .filter(|value| !value.trim().is_empty())
504 .map(str::to_string)
505 })
506 .unwrap_or_else(|| task.task_id.clone());
507 let idempotency_key = format!("{run_id}:cycle:{cycle_index}");
508 let envelope = Self {
509 schema_version: DISTRIBUTED_RUN_SCHEMA_VERSION.to_string(),
510 job_id: idempotency_key.clone(),
511 run_id,
512 task,
513 budget_limits,
514 recipe,
515 cycle_name: cycle_name.into(),
516 cycle_index,
517 idempotency_key,
518 deadline_unix_ms,
519 lease_duration_ms,
520 root_run_id: root_run_id.into(),
521 trace_id: trace_id.into(),
522 run_definition_schema: RUN_DEFINITION_SCHEMA.to_string(),
523 run_definition_digest: run_definition_digest.into(),
524 claim_mode,
525 resume_attempt,
526 checkpoint_config,
527 };
528 envelope.validate()?;
529 Ok(envelope)
530 }
531
532 pub fn validate(&self) -> Result<(), String> {
533 if self.schema_version != DISTRIBUTED_RUN_SCHEMA_VERSION {
534 return Err(format!(
535 "unsupported distributed schema_version: {}",
536 self.schema_version
537 ));
538 }
539 self.validate_current_fields()?;
540 for (field_name, value) in [
541 ("job_id", self.job_id.as_str()),
542 ("run_id", self.run_id.as_str()),
543 ("cycle_name", self.cycle_name.as_str()),
544 ("idempotency_key", self.idempotency_key.as_str()),
545 ] {
546 require_non_empty(value, &format!("distributed envelope {field_name}"))?;
547 }
548 if self.cycle_index == 0 {
549 return Err(
550 "distributed envelope cycle_index must be between 1 and 4294967295".to_string(),
551 );
552 }
553 if self.lease_duration_ms == 0 {
554 return Err(
555 "distributed envelope lease_duration_ms must be a positive integer".to_string(),
556 );
557 }
558 self.recipe.validate()?;
559 if let Some(limits) = &self.budget_limits {
560 limits.validate()?;
561 }
562 Ok(())
563 }
564
565 fn validate_current_fields(&self) -> Result<(), String> {
566 for (field_name, value) in [
567 ("root_run_id", self.root_run_id.as_str()),
568 ("trace_id", self.trace_id.as_str()),
569 ] {
570 require_non_empty(value, &format!("distributed envelope {field_name}"))?;
571 }
572 if self.run_definition_schema != RUN_DEFINITION_SCHEMA {
573 return Err("checkpoint_definition_schema_unsupported".to_string());
574 }
575 validate_sha256(&self.run_definition_digest, "run_definition_digest")
576 .map_err(|error| error.to_string())?;
577 if self.resume_attempt == 0 || self.resume_attempt > MAX_WIRE_INTEGER {
578 return Err("checkpoint_resume_attempt_invalid".to_string());
579 }
580 self.checkpoint_config.validate()?;
581 if self.recipe.capabilities.checkpoint_store_ref.is_none() {
582 return Err("distributed run requires checkpoint_store_ref".to_string());
583 }
584 for namespace in &self.checkpoint_config.required_extension_namespaces {
585 if !self
586 .recipe
587 .capabilities
588 .checkpoint_extension_refs
589 .iter()
590 .any(|reference| reference.namespace == *namespace && reference.required)
591 {
592 return Err(format!(
593 "required checkpoint extension {namespace} is unavailable"
594 ));
595 }
596 }
597 if self
598 .deadline_unix_ms
599 .is_some_and(|value| value > MAX_WIRE_INTEGER)
600 || self.lease_duration_ms > MAX_WIRE_INTEGER
601 {
602 return Err("distributed lease values must be JSON-safe integers".to_string());
603 }
604 Ok(())
605 }
606
607 pub fn ensure_not_expired_at(&self, now_ms: u64) -> Result<(), String> {
608 if self
609 .deadline_unix_ms
610 .is_some_and(|deadline| deadline <= now_ms)
611 {
612 return Err(format!(
613 "distributed job {} deadline has expired",
614 self.job_id
615 ));
616 }
617 Ok(())
618 }
619
620 pub fn ensure_not_expired(&self) -> Result<(), String> {
621 self.ensure_not_expired_at(now_unix_ms()?)
622 }
623
624 pub fn remaining_millis_at(&self, now_ms: u64) -> Option<u64> {
625 self.deadline_unix_ms
626 .map(|deadline| deadline.saturating_sub(now_ms))
627 }
628
629 pub fn to_dict(&self) -> Value {
630 let mut value = serde_json::json!({
631 "schema_version": self.schema_version,
632 "job_id": self.job_id,
633 "run_id": self.run_id,
634 "task": self.task.to_dict(),
635 "budget_limits": self.budget_limits,
636 "recipe": self.recipe.to_dict(),
637 "cycle_name": self.cycle_name,
638 "cycle_index": self.cycle_index,
639 "idempotency_key": self.idempotency_key,
640 "deadline_unix_ms": self.deadline_unix_ms,
641 "lease_duration_ms": self.lease_duration_ms,
642 "root_run_id": self.root_run_id,
643 "trace_id": self.trace_id,
644 "run_definition_schema": self.run_definition_schema,
645 "run_definition_digest": self.run_definition_digest,
646 "claim_mode": self.claim_mode,
647 "resume_attempt": self.resume_attempt,
648 "checkpoint_config": self.checkpoint_config,
649 });
650 let object = value
651 .as_object_mut()
652 .expect("distributed envelope is always an object");
653 normalize_integral_float(
654 object
655 .get_mut("recipe")
656 .and_then(Value::as_object_mut)
657 .expect("runtime recipe is always an object"),
658 "timeout_seconds",
659 );
660 if let Some(capabilities) = object
661 .get_mut("recipe")
662 .and_then(Value::as_object_mut)
663 .and_then(|recipe| recipe.get_mut("capabilities"))
664 .and_then(Value::as_object_mut)
665 {
666 normalize_integral_float(capabilities, "approval_timeout_seconds");
667 }
668 value
669 }
670
671 pub fn from_dict(payload: &Value) -> Result<Self, String> {
672 if !payload.is_object() {
673 return Err("distributed envelope must be an object".to_string());
674 }
675 if let Some(budget_limits) = payload
676 .get("budget_limits")
677 .filter(|budget_limits| !budget_limits.is_null())
678 {
679 serde_json::from_value::<RunBudgetLimits>(budget_limits.clone()).map_err(|error| {
680 format!(
681 "distributed envelope budget limit must be between 0 and {MAX_WIRE_INTEGER}: {error}"
682 )
683 })?;
684 }
685 let schema_version = payload
686 .get("schema_version")
687 .and_then(Value::as_str)
688 .ok_or_else(|| "unsupported distributed schema_version".to_string())?;
689 if schema_version != DISTRIBUTED_RUN_SCHEMA_VERSION {
690 return Err(format!(
691 "unsupported distributed schema_version: {schema_version}"
692 ));
693 }
694 validate_current_discriminator_fields(payload)?;
695 let envelope: Self =
696 serde_json::from_value(payload.clone()).map_err(|error| error.to_string())?;
697 envelope.validate()?;
698 if envelope.to_dict() != *payload {
699 return Err(
700 "distributed envelope must use the complete canonical current wire shape"
701 .to_string(),
702 );
703 }
704 Ok(envelope)
705 }
706}
707
708impl Serialize for DistributedRunEnvelope {
709 fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
710 where
711 S: serde::Serializer,
712 {
713 self.to_dict().serialize(serializer)
714 }
715}