use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::time::Duration;
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;
use tracing::{debug, warn};
use crate::engine::worktrees::{list_worktrees, main_repo_root, wave_name_from_worktree_and_main};
use crate::journal::{events_path, read_events, runs_root};
use crate::lfd::events::EventHub;
use crate::lfd::types::Event;
use crate::lfdb::SharedStore;
const POLL_INTERVAL: Duration = Duration::from_secs(1);
#[derive(Debug, Default)]
struct LfObserver {
line_cursors: HashMap<PathBuf, usize>,
}
impl LfObserver {
async fn scan(&mut self, store: &SharedStore, event_hub: &EventHub) -> Result<(), String> {
let repo_roots = store
.list_waves(None)
.await
.map_err(|err| err.to_string())?
.into_iter()
.map(|wave| PathBuf::from(wave.repo()))
.collect::<HashSet<_>>();
for repo_root in repo_roots {
self.scan_repo(&repo_root, event_hub).await?;
}
Ok(())
}
async fn scan_repo(&mut self, repo_root: &Path, event_hub: &EventHub) -> Result<(), String> {
let main_repo = main_repo_root(repo_root).map_err(|err| err.to_string())?;
let worktrees = tokio::task::spawn_blocking({
let repo = main_repo.clone();
move || list_worktrees(&repo)
})
.await
.map_err(|err| err.to_string())?
.map_err(|err| err.to_string())?;
for worktree in worktrees {
if wave_name_from_worktree_and_main(&worktree.path, &main_repo).is_none() {
continue;
}
self.scan_worktree(&worktree.path, event_hub)?;
}
Ok(())
}
fn scan_worktree(&mut self, worktree: &Path, event_hub: &EventHub) -> Result<(), String> {
let root = runs_root(worktree);
if !root.exists() {
return Ok(());
}
let entries = std::fs::read_dir(&root)
.map_err(|err| format!("failed to read {}: {err}", root.display()))?;
for entry in entries {
let entry = entry.map_err(|err| err.to_string())?;
let path = entry.path();
if !path.is_dir() {
continue;
}
if let Err(err) = self.replay_new_lines(&path, event_hub) {
debug!(error = %err, run_dir = %path.display(), "skipping invalid journal");
}
}
Ok(())
}
fn replay_new_lines(&mut self, run_dir: &Path, event_hub: &EventHub) -> Result<(), String> {
let seen = self.line_cursors.get(run_dir).copied().unwrap_or(0);
let path = events_path(run_dir);
let events = read_events(run_dir)
.map_err(|err| format!("failed reading {}: {err}", path.display()))?;
let processed = events.len();
for event in events.into_iter().skip(seen) {
event_hub.send(Event::from(event));
}
self.line_cursors.insert(run_dir.to_path_buf(), processed);
Ok(())
}
}
pub fn spawn(store: SharedStore, event_hub: EventHub, cancel: CancellationToken) -> JoinHandle<()> {
tokio::spawn(async move {
let mut observer = LfObserver::default();
let mut interval = tokio::time::interval(POLL_INTERVAL);
interval.tick().await;
loop {
tokio::select! {
_ = cancel.cancelled() => break,
_ = interval.tick() => {
if let Err(err) = observer.scan(&store, &event_hub).await {
warn!(error = %err, "journal scan failed");
}
}
}
}
})
}
#[cfg(test)]
mod tests {
use super::LfObserver;
use crate::journal::{emit, LfEventFields, LfEventType, LfNode};
use crate::lfd::events::EventHub;
use loopflow_test_support::TestRepo;
use time::OffsetDateTime;
fn started_fields(
command: &[String],
worktree: &std::path::Path,
wave_name: &str,
) -> LfEventFields {
LfEventFields {
wave_name: Some(wave_name.to_string()),
worktree: Some(worktree.display().to_string()),
command: Some(command.to_vec()),
..LfEventFields::default()
}
}
#[test]
fn observer_replays_new_journal_events() {
let repo = TestRepo::new();
let worktree = repo.create_wave_worktree("observe");
let command = vec!["lf".to_string(), "build".to_string()];
emit(
&worktree,
LfNode::Run,
LfEventType::Started,
started_fields(&command, &worktree, "observe"),
);
emit(
&worktree,
LfNode::Flow,
LfEventType::Started,
LfEventFields {
flow: Some("build".to_string()),
..LfEventFields::default()
},
);
emit(
&worktree,
LfNode::Step,
LfEventType::Started,
LfEventFields {
step: Some("implement".to_string()),
index: Some(0),
..LfEventFields::default()
},
);
emit(
&worktree,
LfNode::Step,
LfEventType::Completed,
LfEventFields {
step: Some("implement".to_string()),
index: Some(0),
..LfEventFields::default()
},
);
emit(
&worktree,
LfNode::Flow,
LfEventType::Completed,
LfEventFields::default(),
);
emit(
&worktree,
LfNode::Run,
LfEventType::Completed,
LfEventFields::default(),
);
let hub = EventHub::new(8);
let mut rx = hub.subscribe();
let mut observer = LfObserver::default();
observer
.scan_worktree(&worktree, &hub)
.expect("scan journal worktree");
let event_types = (0..6)
.map(|_| {
let event = rx.try_recv().expect("event");
let json = serde_json::to_value(&event).expect("serialize");
json["type"].as_str().expect("type").to_string()
})
.collect::<Vec<_>>();
assert_eq!(
event_types,
vec![
"run.started",
"flow.started",
"step.started",
"step.completed",
"flow.completed",
"run.completed",
]
);
observer
.scan_worktree(&worktree, &hub)
.expect("rescan journal worktree");
assert!(
rx.try_recv().is_err(),
"no duplicate events after replay cursor"
);
}
#[test]
fn journal_replay_preserves_original_timestamp() {
let timestamp = OffsetDateTime::parse(
"2026-03-20T18:30:45Z",
&time::format_description::well_known::Rfc3339,
)
.expect("timestamp");
let event = crate::journal::LfEvent {
run_id: "2b2a93a3-59fd-413a-a95a-c9fb41351d4e"
.parse()
.expect("uuid"),
ts: timestamp,
node: LfNode::Run,
event: LfEventType::Completed,
wave_name: None,
worktree: None,
command: None,
flow: None,
step: None,
index: None,
error: None,
signal: None,
};
let mapped = crate::lfd::types::Event::from(event);
let json = serde_json::to_value(&mapped).expect("serialize");
assert_eq!(json["type"], "run.completed");
assert_eq!(json["timestamp"], "2026-03-20T18:30:45Z");
}
}