meerkat_core/

ops.rs

//! Async operation types for Meerkat
//!
//! Unified abstraction for tool calls, shell commands, and delegated branches.

use crate::budget::BudgetLimits;
use crate::session::ToolVisibilityWitness;
use crate::types::Message;
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, BTreeSet};
use uuid::Uuid;

/// Unique identifier for an operation
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct OperationId(pub Uuid);

/// Wait policy for async operations.
///
/// Determines whether an operation blocks the turn boundary (`Barrier`) or runs
/// independently (`Detached`).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum WaitPolicy {
    /// Operation must complete before `ToolCallsResolved` can fire.
    Barrier,
    /// Operation runs independently and does not block the turn.
    Detached,
}

/// Typed async operation reference carrying an operation ID and its wait policy.
///
/// Replaces raw `OperationId` sequences in the TurnExecution machine state to
/// enable barrier-aware scheduling. Only `Barrier` ops block the turn boundary;
/// `Detached` ops are recorded but do not gate `ToolCallsResolved`.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct AsyncOpRef {
    pub operation_id: OperationId,
    pub wait_policy: WaitPolicy,
}

impl WaitPolicy {
    /// Normal tool-call operations that must complete before the turn boundary.
    pub fn barrier() -> Self {
        Self::Barrier
    }

    /// Background or mob-child operations that run independently of the turn.
    pub fn detached() -> Self {
        Self::Detached
    }
}

impl AsyncOpRef {
    /// Create a barrier op ref — blocks the turn boundary until resolved.
    pub fn barrier(operation_id: OperationId) -> Self {
        Self {
            operation_id,
            wait_policy: WaitPolicy::barrier(),
        }
    }

    /// Create a detached op ref — runs independently, does not block the turn.
    pub fn detached(operation_id: OperationId) -> Self {
        Self {
            operation_id,
            wait_policy: WaitPolicy::detached(),
        }
    }
}

/// Outcome of a tool dispatch, separating transcript data from execution metadata.
///
/// `result` is what the model sees (conversation/transcript). `async_ops` is
/// what the runtime scheduler sees (barrier/detached classification). This
/// prevents hooks, persistence, and message serialization from accidentally
/// owning barrier semantics.
/// Typed session-level effect produced by tool dispatch.
///
/// Tools that need to mutate session-owned durable state (e.g., mob authority)
/// must NOT call `SessionService` methods from inside dispatch. Instead they
/// return typed effects here, and the turn owner (agent loop) merges and commits
/// them after the parallel tool batch completes.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "effect_type", rename_all = "snake_case")]
pub enum SessionEffect {
    /// Grant management scope for a specific mob.
    GrantManageMob { mob_id: String },
    /// Record durable deferred-tool requests for subsequent boundaries.
    RequestDeferredTools {
        names: BTreeSet<String>,
        witnesses: BTreeMap<String, ToolVisibilityWitness>,
    },
}

#[derive(Debug, Clone)]
pub struct ToolDispatchOutcome {
    /// The tool result for the conversation transcript.
    pub result: crate::types::ToolResult,
    /// Async operations started by this dispatch, with typed wait policies.
    ///
    /// Empty for synchronous tools. Barrier ops block the turn boundary;
    /// detached ops run independently.
    pub async_ops: Vec<AsyncOpRef>,
    /// Session-level effects to be merged by the turn owner after the batch.
    ///
    /// Most tools return an empty vec. Tools that need durable session state
    /// changes (e.g., mob authority grants) emit typed effects here instead
    /// of calling `SessionService` from inside dispatch.
    pub session_effects: Vec<SessionEffect>,
}

impl ToolDispatchOutcome {
    /// Create an outcome with no async operations or session effects (synchronous tool).
    pub fn sync_result(result: crate::types::ToolResult) -> Self {
        Self {
            result,
            async_ops: Vec::new(),
            session_effects: Vec::new(),
        }
    }
}

impl From<crate::types::ToolResult> for ToolDispatchOutcome {
    fn from(result: crate::types::ToolResult) -> Self {
        Self::sync_result(result)
    }
}

impl OperationId {
    /// Create a new operation ID
    pub fn new() -> Self {
        Self(Uuid::now_v7())
    }
}

impl Default for OperationId {
    fn default() -> Self {
        Self::new()
    }
}

impl std::fmt::Display for OperationId {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.0)
    }
}

/// What kind of work the operation performs
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum WorkKind {
    /// MCP or internal tool call
    ToolCall,
    /// Shell command execution
    ShellCommand,
}

/// Shape of the operation's result
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum ResultShape {
    /// Single result value
    Single,
    /// Streaming output (progress events)
    Stream,
    /// Multiple results (e.g., fork branches)
    Batch,
}

/// How much context a delegated branch receives
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(tag = "type", content = "value", rename_all = "snake_case")]
pub enum ContextStrategy {
    /// Complete conversation history (Fork default)
    #[default]
    FullHistory,
    /// Last N turns from parent
    LastTurns(u32),
    /// Compressed summary of conversation
    Summary { max_tokens: u32 },
    /// Explicit message list
    Custom { messages: Vec<Message> },
}

/// How to allocate budget when forking
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(tag = "type", content = "value", rename_all = "snake_case")]
pub enum ForkBudgetPolicy {
    /// Split remaining budget equally among branches
    #[default]
    Equal,
    /// Split proportionally based on weights
    Proportional,
    /// Fixed budget per branch
    Fixed(u64),
    /// Give all remaining budget to each branch
    Remaining,
}

/// Tool access control for delegated branches
#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type", content = "value", rename_all = "snake_case")]
pub enum ToolAccessPolicy {
    /// Inherit all tools from parent
    #[default]
    Inherit,
    /// Only allow specific tools
    AllowList(Vec<String>),
    /// Block specific tools
    DenyList(Vec<String>),
}

/// Policy for operation execution
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct OperationPolicy {
    /// Timeout for this operation
    pub timeout_ms: Option<u64>,
    /// Whether to cancel on parent cancellation
    pub cancel_on_parent_cancel: bool,
    /// Whether to include in checkpoints
    pub checkpoint_results: bool,
}

/// Complete operation specification
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct OperationSpec {
    pub id: OperationId,
    pub kind: WorkKind,
    pub result_shape: ResultShape,
    pub policy: OperationPolicy,
    pub budget_reservation: BudgetLimits,
    pub depth: u32,
    pub depends_on: Vec<OperationId>,
    pub context: Option<ContextStrategy>,
    pub tool_access: Option<ToolAccessPolicy>,
}

/// Result of a completed operation
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct OperationResult {
    pub id: OperationId,
    pub content: String,
    pub is_error: bool,
    pub duration_ms: u64,
    pub tokens_used: u64,
}

/// Events from operations
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum OpEvent {
    /// Operation started executing
    Started { id: OperationId, kind: WorkKind },

    /// Progress update (for streaming operations)
    Progress {
        id: OperationId,
        message: String,
        percent: Option<f32>,
    },

    /// Operation completed successfully
    Completed {
        id: OperationId,
        result: OperationResult,
    },

    /// Operation failed
    Failed { id: OperationId, error: String },

    /// Operation was cancelled
    Cancelled { id: OperationId },
}

/// Concurrency limits for operations
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConcurrencyLimits {
    /// Maximum delegated-branch nesting depth
    pub max_depth: u32,
    /// Maximum concurrent operations (all types)
    pub max_concurrent_ops: usize,
    /// Maximum concurrent delegated branches specifically
    pub max_concurrent_agents: usize,
    /// Maximum children per parent agent
    pub max_children_per_agent: usize,
}

impl Default for ConcurrencyLimits {
    fn default() -> Self {
        Self {
            max_depth: 3,
            max_concurrent_ops: 32,
            max_concurrent_agents: 8,
            max_children_per_agent: 5,
        }
    }
}

/// Specification for spawning a new delegated branch
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct SpawnSpec {
    /// The prompt/task for the delegated branch
    pub prompt: String,
    /// How much context the delegated branch receives
    pub context: ContextStrategy,
    /// Which tools the delegated branch can access
    pub tool_access: ToolAccessPolicy,
    /// Budget allocation for the delegated branch
    pub budget: BudgetLimits,
    /// If false, the delegated branch cannot spawn/fork further
    pub allow_spawn: bool,
    /// System prompt override (None = inherit from parent)
    pub system_prompt: Option<String>,
}

/// A branch in a fork operation
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ForkBranch {
    /// Identifier for this branch
    pub name: String,
    /// The prompt/task for this branch
    pub prompt: String,
    /// Tool access override (None = inherit)
    pub tool_access: Option<ToolAccessPolicy>,
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
    use super::*;

    #[test]
    fn barrier_constructor_produces_barrier_policy() {
        assert_eq!(WaitPolicy::barrier(), WaitPolicy::Barrier);
        let op_ref = AsyncOpRef::barrier(OperationId::new());
        assert_eq!(op_ref.wait_policy, WaitPolicy::Barrier);
    }

    #[test]
    fn detached_constructor_produces_detached_policy() {
        assert_eq!(WaitPolicy::detached(), WaitPolicy::Detached);
        let op_ref = AsyncOpRef::detached(OperationId::new());
        assert_eq!(op_ref.wait_policy, WaitPolicy::Detached);
    }

    #[test]
    fn test_operation_id_encoding() {
        let id = OperationId::new();
        let json = serde_json::to_string(&id).unwrap();

        let parsed: OperationId = serde_json::from_str(&json).unwrap();
        assert_eq!(id, parsed);
    }

    #[test]
    fn test_work_kind_serialization() {
        assert_eq!(
            serde_json::to_value(WorkKind::ToolCall).unwrap(),
            "tool_call"
        );
        assert_eq!(
            serde_json::to_value(WorkKind::ShellCommand).unwrap(),
            "shell_command"
        );
    }

    #[test]
    fn test_context_strategy_serialization() {
        let full = ContextStrategy::FullHistory;
        let json = serde_json::to_value(&full).unwrap();
        assert_eq!(json["type"], "full_history");

        let last = ContextStrategy::LastTurns(5);
        let json = serde_json::to_value(&last).unwrap();
        assert_eq!(json["type"], "last_turns");
        // Adjacently-tagged: {"type": "last_turns", "value": 5}
        assert_eq!(json["value"], 5);

        let summary = ContextStrategy::Summary { max_tokens: 1000 };
        let json = serde_json::to_value(&summary).unwrap();
        assert_eq!(json["type"], "summary");
        // Adjacently-tagged struct variant: {"type": "summary", "value": {"max_tokens": 1000}}
        assert_eq!(json["value"]["max_tokens"], 1000);

        // Roundtrip
        let parsed: ContextStrategy = serde_json::from_value(json).unwrap();
        match parsed {
            ContextStrategy::Summary { max_tokens } => assert_eq!(max_tokens, 1000),
            _ => unreachable!("Wrong variant"),
        }
    }

    #[test]
    fn test_fork_budget_policy_serialization() {
        let policies = vec![
            (ForkBudgetPolicy::Equal, "equal"),
            (ForkBudgetPolicy::Proportional, "proportional"),
            (ForkBudgetPolicy::Remaining, "remaining"),
        ];

        for (policy, expected_type) in policies {
            let json = serde_json::to_value(&policy).unwrap();
            assert_eq!(json["type"], expected_type);
        }

        let fixed = ForkBudgetPolicy::Fixed(5000);
        let json = serde_json::to_value(&fixed).unwrap();
        assert_eq!(json["type"], "fixed");
        // Adjacently-tagged: {"type": "fixed", "value": 5000}
        assert_eq!(json["value"], 5000);

        // Roundtrip
        let parsed: ForkBudgetPolicy = serde_json::from_value(json).unwrap();
        match parsed {
            ForkBudgetPolicy::Fixed(tokens) => assert_eq!(tokens, 5000),
            _ => unreachable!("Wrong variant"),
        }
    }

    #[test]
    fn test_tool_access_policy_serialization() {
        let inherit = ToolAccessPolicy::Inherit;
        let json = serde_json::to_value(&inherit).unwrap();
        assert_eq!(json["type"], "inherit");

        let allow =
            ToolAccessPolicy::AllowList(vec!["read_file".to_string(), "write_file".to_string()]);
        let json = serde_json::to_value(&allow).unwrap();
        assert_eq!(json["type"], "allow_list");
        // Adjacently-tagged: {"type": "allow_list", "value": [...]}
        assert!(json["value"].is_array());

        let deny = ToolAccessPolicy::DenyList(vec!["dangerous_tool".to_string()]);
        let json = serde_json::to_value(&deny).unwrap();
        assert_eq!(json["type"], "deny_list");
        assert!(json["value"].is_array());

        // Roundtrip
        let parsed: ToolAccessPolicy = serde_json::from_value(json).unwrap();
        match parsed {
            ToolAccessPolicy::DenyList(tools) => {
                assert_eq!(tools.len(), 1);
                assert_eq!(tools[0], "dangerous_tool");
            }
            _ => unreachable!("Wrong variant"),
        }
    }

    #[test]
    fn test_op_event_serialization() {
        let events = vec![
            OpEvent::Started {
                id: OperationId::new(),
                kind: WorkKind::ToolCall,
            },
            OpEvent::Progress {
                id: OperationId::new(),
                message: "50% complete".to_string(),
                percent: Some(0.5),
            },
            OpEvent::Completed {
                id: OperationId::new(),
                result: OperationResult {
                    id: OperationId::new(),
                    content: "result".to_string(),
                    is_error: false,
                    duration_ms: 100,
                    tokens_used: 50,
                },
            },
            OpEvent::Failed {
                id: OperationId::new(),
                error: "timeout".to_string(),
            },
            OpEvent::Cancelled {
                id: OperationId::new(),
            },
        ];

        for event in events {
            let json = serde_json::to_value(&event).unwrap();
            assert!(json.get("type").is_some());

            // Roundtrip
            let _: OpEvent = serde_json::from_value(json).unwrap();
        }
    }

    #[test]
    fn test_concurrency_limits_default() {
        let limits = ConcurrencyLimits::default();
        assert_eq!(limits.max_depth, 3);
        assert_eq!(limits.max_concurrent_ops, 32);
        assert_eq!(limits.max_concurrent_agents, 8);
        assert_eq!(limits.max_children_per_agent, 5);
    }

    #[test]
    fn session_effect_grant_manage_mob_serde_round_trip() {
        let effect = SessionEffect::GrantManageMob {
            mob_id: "test-mob".into(),
        };
        let json = serde_json::to_value(&effect).unwrap();
        let parsed: SessionEffect = serde_json::from_value(json).unwrap();
        assert_eq!(effect, parsed);
    }

    #[test]
    fn tool_dispatch_outcome_with_session_effects() {
        let result = crate::types::ToolResult::new("t1".into(), "ok".into(), false);
        let outcome = ToolDispatchOutcome {
            result,
            async_ops: vec![],
            session_effects: vec![SessionEffect::GrantManageMob {
                mob_id: "mob-1".into(),
            }],
        };
        assert_eq!(outcome.session_effects.len(), 1);
    }

    #[test]
    fn tool_dispatch_outcome_sync_result_has_empty_effects() {
        let result = crate::types::ToolResult::new("t1".into(), "ok".into(), false);
        let outcome = ToolDispatchOutcome::sync_result(result);
        assert!(outcome.session_effects.is_empty());
    }
}