1use crate::budget::BudgetLimits;
6use crate::error::ToolError;
7use crate::session::DeferredToolLoadAuthority;
8use crate::types::{Message, ToolNameSet};
9use serde::{Deserialize, Serialize};
10use uuid::Uuid;
11
12#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
14#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
15pub struct OperationId(#[cfg_attr(feature = "schema", schemars(with = "String"))] pub Uuid);
16
17#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
22#[serde(rename_all = "snake_case")]
23pub enum WaitPolicy {
24 Barrier,
26 Detached,
28}
29
30#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
36pub struct AsyncOpRef {
37 pub operation_id: OperationId,
38 pub wait_policy: WaitPolicy,
39}
40
41impl WaitPolicy {
42 pub fn barrier() -> Self {
44 Self::Barrier
45 }
46
47 pub fn detached() -> Self {
49 Self::Detached
50 }
51}
52
53impl AsyncOpRef {
54 pub fn barrier(operation_id: OperationId) -> Self {
56 Self {
57 operation_id,
58 wait_policy: WaitPolicy::barrier(),
59 }
60 }
61
62 pub fn detached(operation_id: OperationId) -> Self {
64 Self {
65 operation_id,
66 wait_policy: WaitPolicy::detached(),
67 }
68 }
69}
70
71#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
84#[serde(tag = "effect_type", rename_all = "snake_case")]
85pub enum SessionEffect {
86 ReplaceMobToolAuthorityContext {
93 authority_context: crate::service::MobToolAuthorityContext,
94 },
95 RequestDeferredTools {
97 authorities: Vec<DeferredToolLoadAuthority>,
98 },
99 AppendAssistantBlocks {
102 blocks: Vec<crate::types::AssistantBlock>,
103 },
104}
105
106#[derive(Debug, Clone, Copy, PartialEq, Eq)]
107pub enum ToolDispatchTerminalErrorKind {
108 NotFound,
109 Unavailable,
110 InvalidArguments,
111 ExecutionFailed,
112 Timeout,
113 AccessDenied,
114 Other,
115 CallbackPending,
116}
117
118impl From<&ToolError> for ToolDispatchTerminalErrorKind {
119 fn from(error: &ToolError) -> Self {
120 match error {
121 ToolError::NotFound { .. } => Self::NotFound,
122 ToolError::Unavailable { .. } => Self::Unavailable,
123 ToolError::InvalidArguments { .. } => Self::InvalidArguments,
124 ToolError::ExecutionFailed { .. } | ToolError::ExecutionFailedWithData { .. } => {
125 Self::ExecutionFailed
126 }
127 ToolError::Timeout { .. } | ToolError::InactivityTimeout { .. } => Self::Timeout,
128 ToolError::AccessDenied { .. } => Self::AccessDenied,
129 ToolError::Other(_) => Self::Other,
130 ToolError::CallbackPending { .. } => Self::CallbackPending,
131 }
132 }
133}
134
135#[derive(Debug, Clone, PartialEq)]
143pub enum ToolDispatchTerminalCause {
144 RuntimeToolError { error: ToolError },
145}
146
147impl ToolDispatchTerminalCause {
148 #[must_use]
149 pub fn runtime_tool_error(error: &ToolError) -> Self {
150 Self::RuntimeToolError {
151 error: error.clone(),
152 }
153 }
154
155 #[must_use]
157 pub fn kind(&self) -> ToolDispatchTerminalErrorKind {
158 match self {
159 Self::RuntimeToolError { error } => ToolDispatchTerminalErrorKind::from(error),
160 }
161 }
162
163 #[must_use]
169 pub fn to_transcript_content(&self) -> String {
170 match self {
171 Self::RuntimeToolError { error } => error.to_transcript_content(),
172 }
173 }
174
175 #[must_use]
176 pub fn is_runtime_tool_timeout(&self) -> bool {
177 self.kind() == ToolDispatchTerminalErrorKind::Timeout
178 }
179}
180
181#[derive(Debug, Clone)]
182pub struct ToolDispatchOutcome {
183 pub result: crate::types::ToolResult,
185 pub async_ops: Vec<AsyncOpRef>,
190 pub session_effects: Vec<SessionEffect>,
196 terminal_cause: Option<ToolDispatchTerminalCause>,
201}
202
203#[derive(Debug, Clone, Copy, PartialEq, Eq)]
205pub enum ToolDispatchTimeoutPolicy {
206 Default { timeout: std::time::Duration },
208 Disabled,
211 Finite { timeout: std::time::Duration },
213}
214
215impl ToolDispatchTimeoutPolicy {
216 #[must_use]
217 pub fn timeout(self) -> Option<std::time::Duration> {
218 match self {
219 Self::Default { timeout } | Self::Finite { timeout } => Some(timeout),
220 Self::Disabled => None,
221 }
222 }
223
224 #[must_use]
225 pub fn timeout_ms(self) -> Option<u64> {
226 self.timeout()
227 .map(|timeout| u64::try_from(timeout.as_millis()).unwrap_or(u64::MAX))
228 }
229}
230
231impl ToolDispatchOutcome {
232 pub fn new(
234 result: crate::types::ToolResult,
235 async_ops: Vec<AsyncOpRef>,
236 session_effects: Vec<SessionEffect>,
237 ) -> Self {
238 Self {
239 result,
240 async_ops,
241 session_effects,
242 terminal_cause: None,
243 }
244 }
245
246 pub fn sync_result(result: crate::types::ToolResult) -> Self {
248 Self::new(result, Vec::new(), Vec::new())
249 }
250
251 #[must_use]
252 pub fn terminal_cause(&self) -> Option<&ToolDispatchTerminalCause> {
253 self.terminal_cause.as_ref()
254 }
255
256 #[must_use]
257 pub fn is_runtime_tool_timeout(&self) -> bool {
258 self.terminal_cause
259 .as_ref()
260 .is_some_and(ToolDispatchTerminalCause::is_runtime_tool_timeout)
261 }
262
263 pub(crate) fn clear_terminal_cause(&mut self) {
264 self.terminal_cause = None;
265 }
266}
267
268impl From<crate::types::ToolResult> for ToolDispatchOutcome {
269 fn from(result: crate::types::ToolResult) -> Self {
270 Self::sync_result(result)
271 }
272}
273
274pub fn terminal_tool_outcome_for_error(
277 tool_use_id: impl Into<String>,
278 error: ToolError,
279) -> ToolDispatchOutcome {
280 let terminal_cause = ToolDispatchTerminalCause::RuntimeToolError { error };
281 let content = terminal_cause.to_transcript_content();
285 let mut outcome = ToolDispatchOutcome::sync_result(crate::types::ToolResult::new(
286 tool_use_id.into(),
287 content,
288 true,
289 ));
290 outcome.terminal_cause = Some(terminal_cause);
291 outcome
292}
293
294impl OperationId {
295 pub fn new() -> Self {
297 Self(crate::time_compat::new_uuid_v7())
298 }
299
300 pub fn for_detached_job_wait(
307 session_id: &crate::types::SessionId,
308 realm_id: &str,
309 job_id: &str,
310 ) -> Self {
311 let name = format!(
312 "meerkat.detached_job_wait.v1:{}:{}:{}:{}:{}",
313 session_id,
314 realm_id.len(),
315 realm_id,
316 job_id.len(),
317 job_id
318 );
319 Self(Uuid::new_v5(&Uuid::NAMESPACE_URL, name.as_bytes()))
320 }
321}
322
323impl Default for OperationId {
324 fn default() -> Self {
325 Self::new()
326 }
327}
328
329impl std::fmt::Display for OperationId {
330 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
331 write!(f, "{}", self.0)
332 }
333}
334
335#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
337#[serde(rename_all = "snake_case")]
338pub enum WorkKind {
339 ToolCall,
341 ShellCommand,
343}
344
345#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
347#[serde(rename_all = "snake_case")]
348pub enum ResultShape {
349 Single,
351 Stream,
353 Batch,
355}
356
357#[derive(Debug, Clone, Default, Serialize, Deserialize)]
359#[serde(tag = "type", content = "value", rename_all = "snake_case")]
360pub enum ContextStrategy {
361 #[default]
363 FullHistory,
364 LastTurns(u32),
366 Summary { max_tokens: u32 },
368 Custom { messages: Vec<Message> },
370}
371
372#[derive(Debug, Clone, Default, Serialize, Deserialize)]
374#[serde(tag = "type", content = "value", rename_all = "snake_case")]
375pub enum ForkBudgetPolicy {
376 #[default]
378 Equal,
379 Proportional,
381 Fixed(u64),
383 Remaining,
385}
386
387#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
389#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
390#[serde(tag = "type", content = "value", rename_all = "snake_case")]
391pub enum ToolAccessPolicy {
392 #[default]
394 Inherit,
395 AllowList(ToolNameSet),
397 DenyList(ToolNameSet),
399}
400
401#[derive(Debug, Clone, Serialize, Deserialize, Default)]
403pub struct OperationPolicy {
404 pub timeout_ms: Option<u64>,
406 pub cancel_on_parent_cancel: bool,
408 pub checkpoint_results: bool,
410}
411
412#[derive(Debug, Clone, Serialize, Deserialize)]
414pub struct OperationSpec {
415 pub id: OperationId,
416 pub kind: WorkKind,
417 pub result_shape: ResultShape,
418 pub policy: OperationPolicy,
419 pub budget_reservation: BudgetLimits,
420 pub depth: u32,
421 pub depends_on: Vec<OperationId>,
422 pub context: Option<ContextStrategy>,
423 pub tool_access: Option<ToolAccessPolicy>,
424}
425
426#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
428pub struct OperationResult {
429 pub id: OperationId,
430 pub content: String,
431 pub is_error: bool,
432 pub duration_ms: u64,
433 pub tokens_used: u64,
434}
435
436#[derive(Debug, Clone, Serialize, Deserialize)]
438#[serde(tag = "type", rename_all = "snake_case")]
439pub enum OpEvent {
440 Started { id: OperationId, kind: WorkKind },
442
443 Progress {
445 id: OperationId,
446 message: String,
447 percent: Option<f32>,
448 },
449
450 Completed {
452 id: OperationId,
453 result: OperationResult,
454 },
455
456 Failed { id: OperationId, error: String },
458
459 Cancelled { id: OperationId },
461}
462
463#[derive(Debug, Clone, Serialize, Deserialize)]
465pub struct ConcurrencyLimits {
466 pub max_depth: u32,
468 pub max_concurrent_ops: usize,
470 pub max_concurrent_agents: usize,
472 pub max_children_per_agent: usize,
474}
475
476impl Default for ConcurrencyLimits {
477 fn default() -> Self {
478 Self {
479 max_depth: 3,
480 max_concurrent_ops: 32,
481 max_concurrent_agents: 8,
482 max_children_per_agent: 5,
483 }
484 }
485}
486
487#[derive(Debug, Clone, Serialize, Deserialize, Default)]
489pub struct SpawnSpec {
490 pub prompt: String,
492 pub context: ContextStrategy,
494 pub tool_access: ToolAccessPolicy,
496 pub budget: BudgetLimits,
498 pub allow_spawn: bool,
500 pub system_prompt: Option<String>,
502}
503
504#[derive(Debug, Clone, Serialize, Deserialize)]
506pub struct ForkBranch {
507 pub name: String,
509 pub prompt: String,
511 pub tool_access: Option<ToolAccessPolicy>,
513}
514
515#[cfg(test)]
516#[allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)]
517mod tests {
518 use super::*;
519
520 fn generated_mob_authority_for_test() -> crate::service::MobToolAuthorityContext {
521 crate::service::MobToolAuthorityContext::generated_for_test(
522 crate::service::OpaquePrincipalToken::new("generated-effect-test"),
523 true,
524 true,
525 false,
526 std::collections::BTreeSet::from(["test-mob".to_string()]),
527 std::collections::BTreeMap::new(),
528 None,
529 None,
530 )
531 }
532
533 #[test]
534 fn barrier_constructor_produces_barrier_policy() {
535 assert_eq!(WaitPolicy::barrier(), WaitPolicy::Barrier);
536 let op_ref = AsyncOpRef::barrier(OperationId::new());
537 assert_eq!(op_ref.wait_policy, WaitPolicy::Barrier);
538 }
539
540 #[test]
541 fn detached_constructor_produces_detached_policy() {
542 assert_eq!(WaitPolicy::detached(), WaitPolicy::Detached);
543 let op_ref = AsyncOpRef::detached(OperationId::new());
544 assert_eq!(op_ref.wait_policy, WaitPolicy::Detached);
545 }
546
547 #[test]
548 fn test_operation_id_encoding() {
549 let id = OperationId::new();
550 let json = serde_json::to_string(&id).unwrap();
551
552 let parsed: OperationId = serde_json::from_str(&json).unwrap();
553 assert_eq!(id, parsed);
554 }
555
556 #[test]
557 fn test_work_kind_serialization() {
558 assert_eq!(
559 serde_json::to_value(WorkKind::ToolCall).unwrap(),
560 "tool_call"
561 );
562 assert_eq!(
563 serde_json::to_value(WorkKind::ShellCommand).unwrap(),
564 "shell_command"
565 );
566 }
567
568 #[test]
569 fn test_context_strategy_serialization() {
570 let full = ContextStrategy::FullHistory;
571 let json = serde_json::to_value(&full).unwrap();
572 assert_eq!(json["type"], "full_history");
573
574 let last = ContextStrategy::LastTurns(5);
575 let json = serde_json::to_value(&last).unwrap();
576 assert_eq!(json["type"], "last_turns");
577 assert_eq!(json["value"], 5);
579
580 let summary = ContextStrategy::Summary { max_tokens: 1000 };
581 let json = serde_json::to_value(&summary).unwrap();
582 assert_eq!(json["type"], "summary");
583 assert_eq!(json["value"]["max_tokens"], 1000);
585
586 let parsed: ContextStrategy = serde_json::from_value(json).unwrap();
588 match parsed {
589 ContextStrategy::Summary { max_tokens } => assert_eq!(max_tokens, 1000),
590 _ => unreachable!("Wrong variant"),
591 }
592 }
593
594 #[test]
595 fn test_fork_budget_policy_serialization() {
596 let policies = vec![
597 (ForkBudgetPolicy::Equal, "equal"),
598 (ForkBudgetPolicy::Proportional, "proportional"),
599 (ForkBudgetPolicy::Remaining, "remaining"),
600 ];
601
602 for (policy, expected_type) in policies {
603 let json = serde_json::to_value(&policy).unwrap();
604 assert_eq!(json["type"], expected_type);
605 }
606
607 let fixed = ForkBudgetPolicy::Fixed(5000);
608 let json = serde_json::to_value(&fixed).unwrap();
609 assert_eq!(json["type"], "fixed");
610 assert_eq!(json["value"], 5000);
612
613 let parsed: ForkBudgetPolicy = serde_json::from_value(json).unwrap();
615 match parsed {
616 ForkBudgetPolicy::Fixed(tokens) => assert_eq!(tokens, 5000),
617 _ => unreachable!("Wrong variant"),
618 }
619 }
620
621 #[test]
622 fn test_tool_access_policy_serialization() {
623 let inherit = ToolAccessPolicy::Inherit;
624 let json = serde_json::to_value(&inherit).unwrap();
625 assert_eq!(json["type"], "inherit");
626
627 let allow = ToolAccessPolicy::AllowList(["read_file", "write_file"].into_iter().collect());
628 let json = serde_json::to_value(&allow).unwrap();
629 assert_eq!(json["type"], "allow_list");
630 assert!(json["value"].is_array());
632
633 let deny = ToolAccessPolicy::DenyList(["dangerous_tool"].into_iter().collect());
634 let json = serde_json::to_value(&deny).unwrap();
635 assert_eq!(json["type"], "deny_list");
636 assert!(json["value"].is_array());
637
638 let parsed: ToolAccessPolicy = serde_json::from_value(json).unwrap();
640 match parsed {
641 ToolAccessPolicy::DenyList(tools) => {
642 assert_eq!(tools.len(), 1);
643 assert!(tools.contains("dangerous_tool"));
644 }
645 _ => unreachable!("Wrong variant"),
646 }
647 }
648
649 #[test]
650 fn test_op_event_serialization() {
651 let events = vec![
652 OpEvent::Started {
653 id: OperationId::new(),
654 kind: WorkKind::ToolCall,
655 },
656 OpEvent::Progress {
657 id: OperationId::new(),
658 message: "50% complete".to_string(),
659 percent: Some(0.5),
660 },
661 OpEvent::Completed {
662 id: OperationId::new(),
663 result: OperationResult {
664 id: OperationId::new(),
665 content: "result".to_string(),
666 is_error: false,
667 duration_ms: 100,
668 tokens_used: 50,
669 },
670 },
671 OpEvent::Failed {
672 id: OperationId::new(),
673 error: "timeout".to_string(),
674 },
675 OpEvent::Cancelled {
676 id: OperationId::new(),
677 },
678 ];
679
680 for event in events {
681 let json = serde_json::to_value(&event).unwrap();
682 assert!(json.get("type").is_some());
683
684 let _: OpEvent = serde_json::from_value(json).unwrap();
686 }
687 }
688
689 #[test]
690 fn test_concurrency_limits_default() {
691 let limits = ConcurrencyLimits::default();
692 assert_eq!(limits.max_depth, 3);
693 assert_eq!(limits.max_concurrent_ops, 32);
694 assert_eq!(limits.max_concurrent_agents, 8);
695 assert_eq!(limits.max_children_per_agent, 5);
696 }
697
698 #[test]
699 fn session_effect_replace_mob_authority_context_deserializes_without_authority_seal() {
700 let effect = SessionEffect::ReplaceMobToolAuthorityContext {
701 authority_context: generated_mob_authority_for_test(),
702 };
703 let json = serde_json::to_value(&effect).unwrap();
704 let parsed: SessionEffect = serde_json::from_value(json).unwrap();
705 match parsed {
706 SessionEffect::ReplaceMobToolAuthorityContext { authority_context } => {
707 assert!(!authority_context.is_generated_authority_context());
708 assert!(!authority_context.can_create_mobs());
709 assert!(!authority_context.can_mutate_profiles());
710 assert!(!authority_context.can_manage_mob("test-mob"));
711 }
712 other => panic!("unexpected session effect: {other:?}"),
713 }
714 }
715
716 #[test]
717 fn tool_dispatch_outcome_with_session_effects() {
718 let result = crate::types::ToolResult::new("t1".into(), "ok".into(), false);
719 let outcome = ToolDispatchOutcome::new(
720 result,
721 vec![],
722 vec![SessionEffect::ReplaceMobToolAuthorityContext {
723 authority_context: generated_mob_authority_for_test(),
724 }],
725 );
726 assert_eq!(outcome.session_effects.len(), 1);
727 assert_eq!(outcome.terminal_cause(), None);
728 }
729
730 #[test]
731 fn tool_dispatch_outcome_sync_result_has_empty_effects() {
732 let result = crate::types::ToolResult::new("t1".into(), "ok".into(), false);
733 let outcome = ToolDispatchOutcome::sync_result(result);
734 assert!(outcome.session_effects.is_empty());
735 assert_eq!(outcome.terminal_cause(), None);
736 }
737
738 #[test]
739 fn terminal_tool_outcome_carries_runtime_timeout_cause() {
740 let outcome = terminal_tool_outcome_for_error("t1", ToolError::timeout("slow_tool", 50));
741
742 assert!(outcome.result.is_error);
743 assert!(outcome.is_runtime_tool_timeout());
744 let cause = outcome.terminal_cause().expect("terminal cause present");
745 assert_eq!(cause.kind(), ToolDispatchTerminalErrorKind::Timeout);
746 assert_eq!(
747 cause,
748 &ToolDispatchTerminalCause::RuntimeToolError {
749 error: ToolError::timeout("slow_tool", 50),
750 }
751 );
752 }
753
754 #[test]
755 fn terminal_tool_outcome_transcript_text_is_derived_purely_from_terminal_cause() {
756 let error = ToolError::execution_failed_with_data(
760 "boom",
761 serde_json::json!({ "detail": "structured", "n": 7 }),
762 );
763 let outcome = terminal_tool_outcome_for_error("t1", error);
764
765 let cause = outcome.terminal_cause().expect("terminal cause present");
766 assert_eq!(outcome.result.text_content(), cause.to_transcript_content());
767 }
768
769 #[test]
770 fn tool_authored_error_result_has_no_runtime_terminal_cause() {
771 let result =
772 crate::types::ToolResult::new("t1".into(), "{\"error\":\"timeout\"}".into(), true);
773 let outcome = ToolDispatchOutcome::sync_result(result);
774
775 assert!(outcome.result.is_error);
776 assert!(!outcome.is_runtime_tool_timeout());
777 assert_eq!(outcome.terminal_cause(), None);
778 }
779}