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

//! Simplified real-time event streaming support

use anyhow::Result;
use notify::{RecursiveMode, Watcher};
use std::path::PathBuf;
use std::sync::mpsc::channel;
use tokio::fs;
use tokio::io::{AsyncBufReadExt, BufReader};

/// Simple event streamer that watches a file for changes
pub struct SimpleEventStreamer {
    events_file: PathBuf,
}

impl SimpleEventStreamer {
    /// Create a new event streamer
    pub fn new(events_file: PathBuf) -> Self {
        Self { events_file }
    }

    /// Start streaming events with a callback
    pub async fn stream_events<F>(&self, callback: F) -> Result<()>
    where
        F: Fn(String) + Send + 'static,
    {
        let events_file = self.events_file.clone();

        // Track last position in file
        let mut last_pos = 0u64;

        // Set up file watcher
        let (tx, rx) = channel();
        let mut watcher = notify::recommended_watcher(move |res: Result<notify::Event, _>| {
            if let Ok(event) = res {
                let _ = tx.send(event);
            }
        })?;

        // Watch the events file or its parent directory
        let watch_path = if events_file.exists() {
            events_file.clone()
        } else {
            events_file
                .parent()
                .ok_or_else(|| anyhow::anyhow!("Invalid events file path"))?
                .to_path_buf()
        };

        watcher.watch(&watch_path, RecursiveMode::NonRecursive)?;

        // Process file changes
        loop {
            match rx.recv() {
                Ok(_) => {
                    // File changed, read new events
                    if events_file.exists() {
                        last_pos = self.read_new_events(last_pos, &callback).await?;
                    }
                }
                Err(e) => {
                    log::error!("File watcher error: {}", e);
                    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
                }
            }
        }
    }

    /// Read new events from the file
    async fn read_new_events<F>(&self, last_pos: u64, callback: &F) -> Result<u64>
    where
        F: Fn(String),
    {
        use tokio::io::AsyncSeekExt;

        let mut file = fs::File::open(&self.events_file).await?;
        file.seek(tokio::io::SeekFrom::Start(last_pos)).await?;

        let mut reader = BufReader::new(file);
        let mut line = String::new();
        let mut new_pos = last_pos;

        while reader.read_line(&mut line).await? > 0 {
            if !line.trim().is_empty() {
                callback(line.trim().to_string());
            }
            new_pos += line.len() as u64;
            line.clear();
        }

        Ok(new_pos)
    }
}