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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158

//! Stream processor trait and implementations

use super::types::StreamSource;
use anyhow::Result;
use async_trait::async_trait;
use regex::Regex;
use serde_json::Value;
use tokio::sync::mpsc;

/// Trait for processing stream output line by line
#[async_trait]
pub trait StreamProcessor: Send + Sync {
    /// Process a line from the stream
    async fn process_line(&self, line: &str, source: StreamSource) -> Result<()>;

    /// Handle stream completion
    async fn on_complete(&self, exit_code: Option<i32>) -> Result<()>;

    /// Handle stream errors
    async fn on_error(&self, error: &anyhow::Error) -> Result<()>;
}

/// Stream processor for JSON lines
pub struct JsonLineProcessor {
    event_sender: mpsc::Sender<Value>,
    emit_events: bool,
}

impl JsonLineProcessor {
    /// Create a new JSON line processor
    pub fn new(event_sender: mpsc::Sender<Value>, emit_events: bool) -> Self {
        Self {
            event_sender,
            emit_events,
        }
    }
}

#[async_trait]
impl StreamProcessor for JsonLineProcessor {
    async fn process_line(&self, line: &str, _source: StreamSource) -> Result<()> {
        if !self.emit_events {
            return Ok(());
        }

        // Try to parse as JSON
        if let Ok(json) = serde_json::from_str::<Value>(line) {
            // Send the parsed JSON event
            self.event_sender
                .send(json)
                .await
                .map_err(|e| anyhow::anyhow!("Failed to send JSON event: {}", e))?;
        }
        Ok(())
    }

    async fn on_complete(&self, _exit_code: Option<i32>) -> Result<()> {
        // Nothing special to do on completion
        Ok(())
    }

    async fn on_error(&self, error: &anyhow::Error) -> Result<()> {
        tracing::warn!("JSON processor encountered error: {}", error);
        Ok(())
    }
}

/// Stream processor for pattern matching
pub struct PatternMatchProcessor {
    patterns: Vec<Regex>,
    event_sender: mpsc::Sender<PatternMatch>,
}

/// Pattern match event
#[derive(Debug, Clone)]
pub struct PatternMatch {
    pub line: String,
    pub pattern: String,
    pub source: StreamSource,
    pub captures: Vec<String>,
}

impl PatternMatchProcessor {
    /// Create a new pattern match processor
    pub fn new(patterns: Vec<Regex>, event_sender: mpsc::Sender<PatternMatch>) -> Self {
        Self {
            patterns,
            event_sender,
        }
    }
}

#[async_trait]
impl StreamProcessor for PatternMatchProcessor {
    async fn process_line(&self, line: &str, source: StreamSource) -> Result<()> {
        for pattern in &self.patterns {
            if let Some(captures) = pattern.captures(line) {
                let match_event = PatternMatch {
                    line: line.to_string(),
                    pattern: pattern.to_string(),
                    source,
                    captures: captures
                        .iter()
                        .skip(1) // Skip the full match
                        .filter_map(|m| m.map(|s| s.as_str().to_string()))
                        .collect(),
                };

                self.event_sender
                    .send(match_event)
                    .await
                    .map_err(|e| anyhow::anyhow!("Failed to send pattern match event: {}", e))?;
            }
        }
        Ok(())
    }

    async fn on_complete(&self, _exit_code: Option<i32>) -> Result<()> {
        Ok(())
    }

    async fn on_error(&self, error: &anyhow::Error) -> Result<()> {
        tracing::warn!("Pattern processor encountered error: {}", error);
        Ok(())
    }
}

/// Simple logging processor for debugging
pub struct LoggingProcessor {
    prefix: String,
}

impl LoggingProcessor {
    /// Create a new logging processor
    pub fn new(prefix: impl Into<String>) -> Self {
        Self {
            prefix: prefix.into(),
        }
    }
}

#[async_trait]
impl StreamProcessor for LoggingProcessor {
    async fn process_line(&self, line: &str, source: StreamSource) -> Result<()> {
        tracing::debug!("{} [{:?}]: {}", self.prefix, source, line);
        Ok(())
    }

    async fn on_complete(&self, exit_code: Option<i32>) -> Result<()> {
        tracing::debug!("{} completed with exit code: {:?}", self.prefix, exit_code);
        Ok(())
    }

    async fn on_error(&self, error: &anyhow::Error) -> Result<()> {
        tracing::error!("{} error: {}", self.prefix, error);
        Ok(())
    }
}