rsclaw 2026.5.20

AI Agent Engine Compatible with OpenClaw
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

//! Integration test for the FileSource event producer.

use std::time::Duration;

use rsclaw::gateway::watch::source::{FileSource, SourceImpl};
use tokio::sync::{mpsc, oneshot};

#[tokio::test]
async fn file_source_emits_appended_lines() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("app.log");
    tokio::fs::write(&path, "preexisting\n").await.unwrap();

    let (tx, mut rx) = mpsc::channel(64);
    let (stop_tx, stop_rx) = oneshot::channel();
    let src = SourceImpl::File(FileSource { path: path.clone() });

    let handle = tokio::spawn(async move { src.run(tx, stop_rx).await });

    tokio::time::sleep(Duration::from_millis(300)).await;
    tokio::fs::write(&path, "preexisting\nhello\nworld\n")
        .await
        .unwrap();

    let ev1 = tokio::time::timeout(Duration::from_secs(2), rx.recv())
        .await
        .expect("timeout waiting for first event")
        .expect("channel closed");
    assert_eq!(ev1.event, "line");
    assert_eq!(ev1.raw.as_deref(), Some("hello"));

    let ev2 = tokio::time::timeout(Duration::from_secs(2), rx.recv())
        .await
        .expect("timeout for second event")
        .expect("channel closed");
    assert_eq!(ev2.raw.as_deref(), Some("world"));

    let _ = stop_tx.send(());
    let _ = tokio::time::timeout(Duration::from_secs(1), handle).await;
}

#[tokio::test]
async fn file_source_handles_truncation() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("rot.log");
    tokio::fs::write(&path, "old1\nold2\n").await.unwrap();

    let (tx, mut rx) = mpsc::channel(64);
    let (stop_tx, stop_rx) = oneshot::channel();
    let src = SourceImpl::File(FileSource { path: path.clone() });
    let handle = tokio::spawn(async move { src.run(tx, stop_rx).await });

    tokio::time::sleep(Duration::from_millis(300)).await;
    tokio::fs::write(&path, "fresh\n").await.unwrap();

    let ev = tokio::time::timeout(Duration::from_secs(2), rx.recv())
        .await
        .expect("timeout")
        .expect("channel closed");
    assert_eq!(ev.raw.as_deref(), Some("fresh"));

    let _ = stop_tx.send(());
    let _ = tokio::time::timeout(Duration::from_secs(1), handle).await;
}