prodigy 0.4.4

Turn ad-hoc Claude sessions into reproducible development pipelines with parallel AI agents
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

//! Handler command executor for MapReduce operations

use super::executor::{
    CommandError, CommandExecutor, CommandResult, ExecutionContext as MapReduceContext,
};
use crate::commands::{CommandRegistry, ExecutionContext};
use crate::cook::workflow::{CommandType, WorkflowStep};
use async_trait::async_trait;
use std::sync::Arc;
use std::time::Instant;

/// Executor for handler commands
pub struct HandlerCommandExecutor {
    command_registry: Arc<CommandRegistry>,
}

impl HandlerCommandExecutor {
    /// Create a new handler command executor
    pub fn new(command_registry: Arc<CommandRegistry>) -> Self {
        Self { command_registry }
    }

    /// Build execution context for handler
    fn build_exec_context(context: &MapReduceContext) -> ExecutionContext {
        let mut exec_context = ExecutionContext::new(context.worktree_path.clone());

        // Add standard environment variables
        exec_context.add_env_var(
            "PRODIGY_WORKTREE".to_string(),
            context.worktree_name.clone(),
        );
        exec_context.add_env_var("PRODIGY_ITEM_ID".to_string(), context.item_id.clone());
        exec_context.add_env_var("PRODIGY_AUTOMATION".to_string(), "true".to_string());

        // Add context environment variables
        for (key, value) in &context.environment {
            exec_context.add_env_var(key.clone(), value.clone());
        }

        exec_context
    }

    /// Convert JSON attributes to handler attributes
    fn convert_attributes(
        attributes: &std::collections::HashMap<String, serde_json::Value>,
    ) -> std::collections::HashMap<String, crate::commands::AttributeValue> {
        let mut converted = std::collections::HashMap::new();

        for (key, value) in attributes {
            let attr_value = Self::json_to_attribute_value(value);
            converted.insert(key.clone(), attr_value);
        }

        converted
    }

    /// Convert JSON value to AttributeValue
    fn json_to_attribute_value(value: &serde_json::Value) -> crate::commands::AttributeValue {
        match value {
            serde_json::Value::String(s) => crate::commands::AttributeValue::from(s.clone()),
            serde_json::Value::Bool(b) => crate::commands::AttributeValue::from(*b),
            serde_json::Value::Number(n) => {
                if let Some(i) = n.as_i64() {
                    crate::commands::AttributeValue::from(i as i32)
                } else if let Some(f) = n.as_f64() {
                    crate::commands::AttributeValue::from(f)
                } else {
                    crate::commands::AttributeValue::from(n.to_string())
                }
            }
            _ => crate::commands::AttributeValue::from(value.to_string()),
        }
    }

    /// Convert handler result to command result
    fn build_result(result: crate::commands::CommandResult, start: Instant) -> CommandResult {
        let stdout = result.stdout.as_ref().cloned().unwrap_or_else(|| {
            result
                .data
                .as_ref()
                .and_then(|v| v.as_str())
                .unwrap_or("")
                .to_string()
        });

        CommandResult {
            output: Some(stdout),
            exit_code: result.exit_code.unwrap_or(0),
            variables: std::collections::HashMap::new(),
            duration: start.elapsed(),
            success: result.is_success(),
            stderr: result.stderr.unwrap_or_default(),
            json_log_location: result.json_log_location,
        }
    }
}

#[async_trait]
impl CommandExecutor for HandlerCommandExecutor {
    async fn execute(
        &self,
        step: &WorkflowStep,
        context: &MapReduceContext,
    ) -> Result<CommandResult, CommandError> {
        let start = Instant::now();

        // Extract handler information
        let handler = step
            .handler
            .as_ref()
            .ok_or_else(|| CommandError::InvalidConfiguration("No handler in step".to_string()))?;

        // Build execution context and convert attributes
        let exec_context = Self::build_exec_context(context);
        let converted_attributes = Self::convert_attributes(&handler.attributes);

        // Execute the handler
        let result = self
            .command_registry
            .execute(&handler.name, &exec_context, converted_attributes)
            .await;

        // Convert to command result
        Ok(Self::build_result(result, start))
    }

    fn supports(&self, command_type: &CommandType) -> bool {
        matches!(command_type, CommandType::Handler { .. })
    }
}