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
131
132
133
134
135
136
137
138
139

//! Tests for Claude streaming functionality

#[cfg(test)]
mod tests {
    use crate::cook::execution::claude::{ClaudeExecutor, ClaudeExecutorImpl};
    use crate::cook::execution::runner::tests::MockCommandRunner;
    use crate::cook::execution::ExecutionResult;
    use std::collections::HashMap;
    use std::path::Path;

    #[tokio::test]
    async fn test_claude_streaming_mode_enabled() {
        let mock_runner = MockCommandRunner::new();

        // Simulate Claude streaming JSON output
        let streaming_output = r#"{"event": "session_started", "session_id": "sess_123", "model": "claude-3"}
{"event": "tool_use", "tool_name": "Read", "tool_id": "tool_456", "parameters": {"file": "test.rs"}}
{"event": "message", "content": "Processing file...", "type": "text"}
{"event": "token_usage", "input_tokens": 100, "output_tokens": 50, "cache_read_tokens": 25}
Result: File processed successfully"#;

        mock_runner.add_response(ExecutionResult {
            success: true,
            stdout: streaming_output.to_string(),
            stderr: String::new(),
            exit_code: Some(0),
            metadata: HashMap::new(),
        });

        let executor = ClaudeExecutorImpl::new(mock_runner);
        let mut env_vars = HashMap::new();
        env_vars.insert("PRODIGY_CLAUDE_STREAMING".to_string(), "true".to_string());

        let result = executor
            .execute_claude_command("/test-command", Path::new("/tmp"), env_vars)
            .await
            .unwrap();

        assert!(result.success);
        assert!(result.stdout.contains("session_started"));
        assert!(result.stdout.contains("tool_use"));
        assert!(result.stdout.contains("File processed successfully"));
    }

    #[tokio::test]
    async fn test_claude_streaming_mode_disabled() {
        let mock_runner = MockCommandRunner::new();

        mock_runner.add_response(ExecutionResult {
            success: true,
            stdout: "Command executed in print mode".to_string(),
            stderr: String::new(),
            exit_code: Some(0),
            metadata: HashMap::new(),
        });

        let executor = ClaudeExecutorImpl::new(mock_runner);
        let mut env_vars = HashMap::new();
        env_vars.insert("PRODIGY_CLAUDE_STREAMING".to_string(), "false".to_string());

        let result = executor
            .execute_claude_command("/test-command", Path::new("/tmp"), env_vars)
            .await
            .unwrap();

        assert!(result.success);
        assert_eq!(result.stdout, "Command executed in print mode");
    }

    #[tokio::test]
    async fn test_claude_streaming_fallback_on_error() {
        let mock_runner = MockCommandRunner::new();

        // Simulate mixed JSON and non-JSON output
        let mixed_output = r#"{"event": "session_started", "session_id": "sess_123"}
ERROR: Something went wrong
{"event": "message", "content": "Recovering...", "type": "text"}
Done"#;

        mock_runner.add_response(ExecutionResult {
            success: false,
            stdout: mixed_output.to_string(),
            stderr: "Error occurred".to_string(),
            exit_code: Some(1),
            metadata: HashMap::new(),
        });

        let executor = ClaudeExecutorImpl::new(mock_runner);
        let mut env_vars = HashMap::new();
        env_vars.insert("PRODIGY_CLAUDE_STREAMING".to_string(), "true".to_string());

        let result = executor
            .execute_claude_command("/test-command", Path::new("/tmp"), env_vars)
            .await;

        // With enhanced error handling, failed Claude commands now return Err
        assert!(result.is_err());
        let error = result.unwrap_err();
        assert!(error
            .to_string()
            .contains("Claude command '/test-command' failed"));
        assert!(error.to_string().contains("stderr: Error occurred"));
    }

    #[tokio::test]
    async fn test_claude_processor_json_parsing() {
        use crate::subprocess::streaming::{
            ClaudeJsonProcessor, LoggingClaudeHandler, StreamProcessor, StreamSource,
        };
        use std::sync::Arc;

        let handler = Arc::new(LoggingClaudeHandler::new("test"));
        let processor = ClaudeJsonProcessor::new(handler, false);

        // Test various Claude event types
        let events = vec![
            r#"{"event": "tool_use", "tool_name": "Bash", "tool_id": "123", "parameters": {"command": "ls"}}"#,
            r#"{"event": "token_usage", "input_tokens": 200, "output_tokens": 100, "cache_read_tokens": 50}"#,
            r#"{"event": "message", "content": "Hello", "type": "text"}"#,
            r#"{"event": "session_started", "session_id": "sess_456", "model": "claude-3", "tools": ["Read", "Write"]}"#,
            r#"Regular text output line"#,
        ];

        for event in events {
            processor
                .process_line(event, StreamSource::Stdout)
                .await
                .unwrap();
        }

        // Verify buffer accumulation
        let buffer = processor.get_buffer().await;
        assert!(buffer.contains("tool_use"));
        assert!(buffer.contains("token_usage"));
        assert!(buffer.contains("message"));
        assert!(buffer.contains("session_started"));
        assert!(buffer.contains("Regular text output"));
    }
}