use std::collections::BTreeMap;
use std::path::PathBuf;
use std::sync::{Arc, Mutex, RwLock};
use std::time::Duration;
use notify::{Event, RecommendedWatcher, RecursiveMode, Watcher};
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tokio::time::{Instant, MissedTickBehavior};
use tracing::{info, warn};
use crate::domain::{FriggError, FriggResult};
use crate::indexer::{
IndexMode, IndexPlan, IndexProgressEvent, IndexSummary,
index_repository_with_runtime_config_and_dirty_paths_and_progress_callback,
};
use crate::manifest_validation::ValidatedManifestCandidateCache;
use crate::mcp::types::{RuntimeTaskKind, RuntimeTaskStatus};
use crate::mcp::workspace_registry::AttachedWorkspace;
use crate::mcp::{RuntimeTaskGuard, RuntimeTaskRegistry};
use crate::settings::{
FriggConfig, RuntimeTransportKind, SemanticRuntimeConfig, SemanticRuntimeCredentials,
};
use super::repository::{
WatchedRepository, event_kind_is_relevant, repository_id_for_path,
repository_relative_watch_path, should_ignore_watch_path, startup_refresh_status,
watched_repository_for_workspace,
};
use super::scheduler::{ScheduledRefresh, WatchRefreshClass, WatchSchedulerState};
const WATCH_TICK_MS: u64 = 50;
pub type RepositoryCacheInvalidationCallback = Arc<dyn Fn(&str) + Send + Sync + 'static>;
pub type WatchEventReporter = Arc<dyn Fn(WatchEvent) + Send + Sync + 'static>;
#[derive(Debug, Clone)]
pub enum WatchEvent {
RuntimeStarted {
watch_mode: String,
transport: RuntimeTransportKind,
debounce_ms: u64,
retry_ms: u64,
},
RuntimeDisabled {
watch_mode: String,
transport: RuntimeTransportKind,
},
LeaseAcquired {
repository_id: String,
lease_count: usize,
root: PathBuf,
},
LeaseReleased {
repository_id: String,
remaining: usize,
root: Option<PathBuf>,
},
PathAccepted {
repository_id: String,
path: PathBuf,
},
StartupFresh {
repository_id: String,
root: PathBuf,
snapshot_id: Option<String>,
},
RefreshQueued {
repository_id: String,
root: PathBuf,
refresh_class: &'static str,
reason: String,
snapshot_id: Option<String>,
debounce_ms: Option<u64>,
},
RefreshStarting {
repository_id: String,
root: PathBuf,
refresh_class: &'static str,
debounce_ms: u64,
sampled_paths: usize,
},
RefreshDeferred {
repository_id: String,
root: PathBuf,
refresh_class: &'static str,
reason: &'static str,
retry_ms: u128,
},
IndexPlan {
repository_id: String,
refresh_class: &'static str,
plan: IndexPlan,
},
IndexProgress {
repository_id: String,
refresh_class: &'static str,
progress: IndexProgressEvent,
},
RefreshSucceeded {
repository_id: String,
root: PathBuf,
refresh_class: &'static str,
summary: IndexSummary,
},
RefreshFailed {
repository_id: String,
root: PathBuf,
refresh_class: &'static str,
retry_ms: u128,
error: String,
},
RefreshBlocked {
repository_id: String,
root: PathBuf,
refresh_class: &'static str,
reason: &'static str,
error: String,
},
StaleCompletionIgnored {
repository_id: String,
refresh_class: &'static str,
completion_epoch: u64,
current_epoch: u64,
},
NotifyDropped {
error: String,
},
}
fn report_watch_event(reporter: &Option<WatchEventReporter>, event: WatchEvent) {
if let Some(reporter) = reporter {
reporter(event);
}
}
enum SupervisorCommand {
Event(Event),
NotifyDropped {
error: String,
},
LeaseAcquired {
repository: WatchedRepository,
},
LeaseReleased {
repository_id: String,
},
IndexCompleted {
repository_id: String,
epoch: u64,
class: WatchRefreshClass,
result: Result<crate::indexer::IndexSummary, WatchRefreshFailure>,
},
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum WatchRefreshFailureKind {
Retryable,
StorageSchemaIncompatible,
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct WatchRefreshFailure {
message: String,
kind: WatchRefreshFailureKind,
}
impl WatchRefreshFailure {
fn from_error(error: FriggError) -> Self {
let kind = if matches!(&error, FriggError::StorageSchemaIncompatible { .. }) {
WatchRefreshFailureKind::StorageSchemaIncompatible
} else {
WatchRefreshFailureKind::Retryable
};
Self {
message: error.to_string(),
kind,
}
}
fn blocks_retry(&self) -> bool {
self.kind == WatchRefreshFailureKind::StorageSchemaIncompatible
}
}
#[derive(Debug, Default)]
struct RepositoryEpochs {
epochs: BTreeMap<String, u64>,
}
impl RepositoryEpochs {
fn ensure(&mut self, repository_id: &str) {
self.epochs.entry(repository_id.to_owned()).or_insert(0);
}
fn bump(&mut self, repository_id: &str) {
self.epochs
.entry(repository_id.to_owned())
.and_modify(|epoch| *epoch = epoch.wrapping_add(1))
.or_insert(0);
}
fn current(&self, repository_id: &str) -> u64 {
self.epochs.get(repository_id).copied().unwrap_or(0)
}
}
#[derive(Debug, Clone, Copy, Default)]
pub struct WatchLeaseStatus {
pub active: bool,
pub lease_count: usize,
}
pub struct WatchRuntime {
watcher: Mutex<RecommendedWatcher>,
repositories: Arc<RwLock<BTreeMap<String, WatchedRepository>>>,
lease_counts: Arc<RwLock<BTreeMap<String, usize>>>,
supervisor_handle: JoinHandle<()>,
command_tx: mpsc::UnboundedSender<SupervisorCommand>,
reporter: Option<WatchEventReporter>,
}
impl WatchRuntime {
pub(crate) fn acquire_lease(&self, workspace: &AttachedWorkspace) -> FriggResult<usize> {
let repository = watched_repository_for_workspace(workspace)?;
let lease_count = {
let mut lease_counts = self
.lease_counts
.write()
.unwrap_or_else(|poisoned| poisoned.into_inner());
if let Some(count) = lease_counts.get_mut(&repository.repository_id)
&& *count > 0
{
*count = count.saturating_add(1);
*count
} else {
self.watcher
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.watch(&repository.root, RecursiveMode::Recursive)
.map_err(|err| {
FriggError::Internal(format!(
"failed to register watcher for root {}: {err}",
repository.root.display()
))
})?;
lease_counts.insert(repository.repository_id.clone(), 1);
1
}
};
if lease_count > 1 {
report_watch_event(
&self.reporter,
WatchEvent::LeaseAcquired {
repository_id: repository.repository_id.clone(),
lease_count,
root: repository.root.clone(),
},
);
return Ok(lease_count);
}
self.repositories
.write()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.insert(repository.repository_id.clone(), repository.clone());
let event = WatchEvent::LeaseAcquired {
repository_id: repository.repository_id.clone(),
lease_count,
root: repository.root.clone(),
};
let _ = self
.command_tx
.send(SupervisorCommand::LeaseAcquired { repository });
report_watch_event(&self.reporter, event);
Ok(lease_count)
}
pub(crate) fn release_lease(&self, repository_id: &str) -> usize {
let root = self
.repositories
.read()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.get(repository_id)
.map(|repository| repository.root.clone());
let remaining = {
let mut lease_counts = self
.lease_counts
.write()
.unwrap_or_else(|poisoned| poisoned.into_inner());
let Some(count) = lease_counts.get_mut(repository_id) else {
return 0;
};
*count = count.saturating_sub(1);
let remaining = *count;
if remaining == 0 {
lease_counts.remove(repository_id);
}
remaining
};
if remaining > 0 {
report_watch_event(
&self.reporter,
WatchEvent::LeaseReleased {
repository_id: repository_id.to_owned(),
remaining,
root,
},
);
return remaining;
}
if let Some(repository) = self
.repositories
.write()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.remove(repository_id)
&& let Err(error) = self
.watcher
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.unwatch(&repository.root)
{
warn!(
repository_id,
root = %repository.root.display(),
error = %error,
"built-in watch mode failed to unregister workspace root"
);
}
let _ = self.command_tx.send(SupervisorCommand::LeaseReleased {
repository_id: repository_id.to_owned(),
});
report_watch_event(
&self.reporter,
WatchEvent::LeaseReleased {
repository_id: repository_id.to_owned(),
remaining,
root,
},
);
remaining
}
pub(crate) fn lease_status(&self, repository_id: &str) -> WatchLeaseStatus {
let lease_count = self
.lease_counts
.read()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.get(repository_id)
.copied()
.unwrap_or(0);
WatchLeaseStatus {
active: lease_count > 0,
lease_count,
}
}
#[cfg(test)]
pub(crate) fn inject_test_event(&self, event: Event) {
let _ = self.command_tx.send(SupervisorCommand::Event(event));
}
#[cfg(test)]
pub(crate) fn inject_test_notify_dropped(&self, error: impl Into<String>) {
let _ = self.command_tx.send(SupervisorCommand::NotifyDropped {
error: error.into(),
});
}
}
impl Drop for WatchRuntime {
fn drop(&mut self) {
self.supervisor_handle.abort();
}
}
pub fn maybe_start_watch_runtime(
config: &FriggConfig,
transport: RuntimeTransportKind,
task_registry: Arc<RwLock<RuntimeTaskRegistry>>,
validated_manifest_candidate_cache: Arc<RwLock<ValidatedManifestCandidateCache>>,
repository_cache_invalidation_callback: Option<RepositoryCacheInvalidationCallback>,
) -> FriggResult<Option<WatchRuntime>> {
maybe_start_watch_runtime_with_reporter(
config,
transport,
task_registry,
validated_manifest_candidate_cache,
repository_cache_invalidation_callback,
None,
)
}
pub fn maybe_start_watch_runtime_with_reporter(
config: &FriggConfig,
transport: RuntimeTransportKind,
task_registry: Arc<RwLock<RuntimeTaskRegistry>>,
validated_manifest_candidate_cache: Arc<RwLock<ValidatedManifestCandidateCache>>,
repository_cache_invalidation_callback: Option<RepositoryCacheInvalidationCallback>,
reporter: Option<WatchEventReporter>,
) -> FriggResult<Option<WatchRuntime>> {
config.watch.validate()?;
if !config.watch.enabled_for_transport(transport) {
info!(
watch_mode = %config.watch.mode.as_str(),
transport = ?transport,
"built-in watch mode disabled for resolved transport"
);
report_watch_event(
&reporter,
WatchEvent::RuntimeDisabled {
watch_mode: config.watch.mode.as_str().to_owned(),
transport,
},
);
return Ok(None);
}
let repositories = Arc::new(RwLock::new(BTreeMap::new()));
let lease_counts = Arc::new(RwLock::new(BTreeMap::new()));
let (command_tx, command_rx) = mpsc::unbounded_channel();
let callback_tx = command_tx.clone();
let watcher = notify::recommended_watcher(move |result: notify::Result<Event>| match result {
Ok(event) => {
let _ = callback_tx.send(SupervisorCommand::Event(event));
}
Err(error) => {
let _ = callback_tx.send(SupervisorCommand::NotifyDropped {
error: error.to_string(),
});
}
})
.map_err(|err| FriggError::Internal(format!("failed to create filesystem watcher: {err}")))?;
let watch_config = config.watch.clone();
let semantic_runtime = config.semantic_runtime.clone();
let semantic_credentials = SemanticRuntimeCredentials::from_process_env();
let supervisor_handle = tokio::spawn(run_supervisor(
Arc::clone(&repositories),
watch_config,
semantic_runtime,
semantic_credentials,
task_registry,
validated_manifest_candidate_cache,
repository_cache_invalidation_callback,
reporter.clone(),
command_rx,
command_tx.clone(),
));
report_watch_event(
&reporter,
WatchEvent::RuntimeStarted {
watch_mode: config.watch.mode.as_str().to_owned(),
transport,
debounce_ms: config.watch.debounce_ms,
retry_ms: config.watch.retry_ms,
},
);
Ok(Some(WatchRuntime {
watcher: Mutex::new(watcher),
repositories,
lease_counts,
supervisor_handle,
command_tx,
reporter,
}))
}
#[allow(clippy::too_many_arguments)]
async fn run_supervisor(
repositories: Arc<RwLock<BTreeMap<String, WatchedRepository>>>,
watch_config: crate::settings::WatchConfig,
semantic_runtime: SemanticRuntimeConfig,
semantic_credentials: SemanticRuntimeCredentials,
task_registry: Arc<RwLock<RuntimeTaskRegistry>>,
validated_manifest_candidate_cache: Arc<RwLock<ValidatedManifestCandidateCache>>,
repository_cache_invalidation_callback: Option<RepositoryCacheInvalidationCallback>,
reporter: Option<WatchEventReporter>,
mut command_rx: mpsc::UnboundedReceiver<SupervisorCommand>,
command_tx: mpsc::UnboundedSender<SupervisorCommand>,
) {
let debounce = Duration::from_millis(watch_config.debounce_ms);
let retry = Duration::from_millis(watch_config.retry_ms);
let mut scheduler = WatchSchedulerState::with_concurrency_limits(
0,
watch_config.manifest_fast_concurrency,
watch_config.semantic_followup_concurrency,
);
let mut repository_epochs = RepositoryEpochs::default();
let mut ticker = tokio::time::interval(Duration::from_millis(WATCH_TICK_MS));
ticker.set_missed_tick_behavior(MissedTickBehavior::Skip);
loop {
tokio::select! {
maybe_command = command_rx.recv() => {
let Some(command) = maybe_command else {
break;
};
let now = Instant::now();
match command {
SupervisorCommand::Event(event) => handle_notify_event(
&repositories,
&mut scheduler,
&validated_manifest_candidate_cache,
repository_cache_invalidation_callback.as_ref(),
event,
now,
debounce,
&reporter,
),
SupervisorCommand::NotifyDropped { error } => handle_notify_dropped(
&repositories,
&mut scheduler,
&validated_manifest_candidate_cache,
repository_cache_invalidation_callback.as_ref(),
error,
now,
debounce,
&reporter,
),
SupervisorCommand::LeaseAcquired { repository } => {
repository_epochs.ensure(&repository.repository_id);
scheduler.add_repository(&repository.repository_id);
queue_startup_refresh_if_needed(
&repository,
&mut scheduler,
now,
&semantic_runtime,
&semantic_credentials,
watch_config.debounce_ms,
&reporter,
);
}
SupervisorCommand::LeaseReleased { repository_id } => {
repository_epochs.bump(&repository_id);
scheduler.remove_repository(&repository_id);
}
SupervisorCommand::IndexCompleted {
repository_id,
epoch,
class,
result,
} => {
let current_epoch = repository_epochs.current(&repository_id);
if epoch != current_epoch {
warn!(
repository_id = %repository_id,
refresh_class = %class.as_str(),
completion_epoch = epoch,
current_epoch,
"built-in watch mode ignoring stale index completion"
);
report_watch_event(
&reporter,
WatchEvent::StaleCompletionIgnored {
repository_id: repository_id.clone(),
refresh_class: class.as_str(),
completion_epoch: epoch,
current_epoch,
},
);
invalidate_caches_for_stale_completion(
&repository_id,
&result,
repository_cache_invalidation_callback.as_ref(),
);
} else {
handle_index_completed(
&repositories,
&mut scheduler,
&repository_id,
class,
result,
repository_cache_invalidation_callback.as_ref(),
now,
retry,
&semantic_runtime,
&semantic_credentials,
&reporter,
);
}
}
}
}
_ = ticker.tick() => {}
}
let now = Instant::now();
if let Some(ScheduledRefresh {
repository_id,
class,
..
}) = scheduler.next_ready_refresh(now)
{
let repository = repositories
.read()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.get(&repository_id)
.cloned();
let Some(repository) = repository else {
continue;
};
let stable_repository_id =
crate::domain::model::stable_repository_id_for_root(&repository.root).0;
let repository_ids = watch_runtime_task_repository_aliases(
&repository.repository_id,
&stable_repository_id,
);
let repository_id_refs = repository_ids
.iter()
.map(String::as_str)
.collect::<Vec<_>>();
let task_guard = RuntimeTaskGuard::try_start_if_no_active_for_any_repository(
Arc::clone(&task_registry),
conflicting_task_kinds_for_class(class),
&repository_id_refs,
watch_task_kind_for_class(class),
repository.repository_id.clone(),
watch_task_phase_for_class(class),
Some(format!(
"watch root {} class {}",
repository.root.display(),
class.as_str()
)),
);
let mut task_guard = match task_guard {
Ok(task_guard) => task_guard,
Err(_) => {
let retry_delay = scheduler
.mark_failed(&repository_id, class, now, retry)
.unwrap_or(retry);
report_watch_event(
&reporter,
WatchEvent::RefreshDeferred {
repository_id: repository.repository_id.clone(),
root: repository.root.clone(),
refresh_class: class.as_str(),
reason: "active_task",
retry_ms: retry_delay.as_millis(),
},
);
continue;
}
};
let recent_paths = scheduler.mark_started(&repository_id, class);
let dispatch_epoch = repository_epochs.current(&repository.repository_id);
info!(
repository_id = %repository.repository_id,
root = %repository.root.display(),
refresh_class = %class.as_str(),
debounce_ms = watch_config.debounce_ms,
"built-in watch mode starting refresh"
);
report_watch_event(
&reporter,
WatchEvent::RefreshStarting {
repository_id: repository.repository_id.clone(),
root: repository.root.clone(),
refresh_class: class.as_str(),
debounce_ms: watch_config.debounce_ms,
sampled_paths: recent_paths.len(),
},
);
let completion_tx = command_tx.clone();
let semantic_runtime = semantic_runtime.clone();
let semantic_credentials = semantic_credentials.clone();
let validated_manifest_candidate_cache =
Arc::clone(&validated_manifest_candidate_cache);
let repository_cache_invalidation_callback =
repository_cache_invalidation_callback.clone();
let reporter = reporter.clone();
tokio::task::spawn_blocking(move || {
let result = match class {
WatchRefreshClass::ManifestFast => {
let mut lexical_only_runtime = semantic_runtime.clone();
lexical_only_runtime.enabled = false;
let plan_reporter = reporter.clone();
let progress_reporter = reporter.clone();
let plan_repository_id = repository.repository_id.clone();
let progress_repository_id = repository.repository_id.clone();
index_repository_with_runtime_config_and_dirty_paths_and_progress_callback(
&repository.repository_id,
&repository.root,
&repository.db_path,
IndexMode::ChangedOnly,
&lexical_only_runtime,
&semantic_credentials,
&recent_paths,
move |plan| {
report_watch_event(
&plan_reporter,
WatchEvent::IndexPlan {
repository_id: plan_repository_id.clone(),
refresh_class: class.as_str(),
plan: plan.clone(),
},
);
Ok(())
},
move |progress| {
report_watch_event(
&progress_reporter,
WatchEvent::IndexProgress {
repository_id: progress_repository_id.clone(),
refresh_class: class.as_str(),
progress,
},
);
},
)
}
WatchRefreshClass::SemanticFollowup => {
let plan_reporter = reporter.clone();
let progress_reporter = reporter.clone();
let plan_repository_id = repository.repository_id.clone();
let progress_repository_id = repository.repository_id.clone();
let index_mode = if recent_paths.is_empty() {
IndexMode::Full
} else {
IndexMode::ChangedOnly
};
index_repository_with_runtime_config_and_dirty_paths_and_progress_callback(
&repository.repository_id,
&repository.root,
&repository.db_path,
index_mode,
&semantic_runtime,
&semantic_credentials,
&recent_paths,
move |plan| {
report_watch_event(
&plan_reporter,
WatchEvent::IndexPlan {
repository_id: plan_repository_id.clone(),
refresh_class: class.as_str(),
plan: plan.clone(),
},
);
Ok(())
},
move |progress| {
report_watch_event(
&progress_reporter,
WatchEvent::IndexProgress {
repository_id: progress_repository_id.clone(),
refresh_class: class.as_str(),
progress,
},
);
},
)
}
}
.map_err(WatchRefreshFailure::from_error);
let detail = match &result {
Ok(summary) => Some(format!(
"refresh_class={} snapshot={} files_changed={} files_deleted={} changed_paths={} deleted_paths={} duration_ms={}",
class.as_str(),
summary.snapshot_id,
summary.files_changed,
summary.files_deleted,
summary.changed_paths.len(),
summary.deleted_paths.len(),
summary.duration_ms
)),
Err(err) => Some(format!(
"refresh_class={} error={}",
class.as_str(),
err.message
)),
};
let status = if result.is_ok() {
RuntimeTaskStatus::Succeeded
} else {
RuntimeTaskStatus::Failed
};
if result.is_ok() && class == WatchRefreshClass::ManifestFast {
validated_manifest_candidate_cache
.write()
.expect("validated manifest candidate cache poisoned")
.invalidate_root(&repository.root);
}
if let Some(callback) = repository_cache_invalidation_callback.as_ref() {
callback(&repository.repository_id);
}
task_guard.finish(status, detail);
let _ = completion_tx.send(SupervisorCommand::IndexCompleted {
repository_id: repository.repository_id.clone(),
epoch: dispatch_epoch,
class,
result,
});
});
}
}
}
fn watch_task_kind_for_class(class: WatchRefreshClass) -> RuntimeTaskKind {
match class {
WatchRefreshClass::ManifestFast => RuntimeTaskKind::ChangedIndex,
WatchRefreshClass::SemanticFollowup => RuntimeTaskKind::SemanticRefresh,
}
}
fn conflicting_task_kinds_for_class(class: WatchRefreshClass) -> &'static [RuntimeTaskKind] {
match class {
WatchRefreshClass::ManifestFast => &[
RuntimeTaskKind::ChangedIndex,
RuntimeTaskKind::WorkspaceIndex,
RuntimeTaskKind::WorkspacePrepare,
],
WatchRefreshClass::SemanticFollowup => &[
RuntimeTaskKind::SemanticRefresh,
RuntimeTaskKind::WorkspaceIndex,
RuntimeTaskKind::WorkspacePrepare,
],
}
}
#[cfg(test)]
fn active_refresh_task_running_for_repository(
registry: &RuntimeTaskRegistry,
class: WatchRefreshClass,
repository_id: &str,
stable_repository_id: &str,
) -> bool {
let repository_ids = watch_runtime_task_repository_aliases(repository_id, stable_repository_id);
let repository_ids = repository_ids
.iter()
.map(String::as_str)
.collect::<Vec<_>>();
!registry
.active_tasks_for_kinds_and_any_repository(
conflicting_task_kinds_for_class(class),
&repository_ids,
)
.is_empty()
}
fn watch_runtime_task_repository_aliases(
repository_id: &str,
stable_repository_id: &str,
) -> Vec<String> {
if repository_id == stable_repository_id {
vec![repository_id.to_owned()]
} else {
vec![repository_id.to_owned(), stable_repository_id.to_owned()]
}
}
fn watch_task_phase_for_class(class: WatchRefreshClass) -> &'static str {
match class {
WatchRefreshClass::ManifestFast => "watch_manifest_fast",
WatchRefreshClass::SemanticFollowup => "watch_semantic_followup",
}
}
#[allow(clippy::too_many_arguments)]
fn handle_notify_event(
repositories: &Arc<RwLock<BTreeMap<String, WatchedRepository>>>,
scheduler: &mut WatchSchedulerState,
validated_manifest_candidate_cache: &Arc<RwLock<ValidatedManifestCandidateCache>>,
repository_cache_invalidation_callback: Option<&RepositoryCacheInvalidationCallback>,
event: Event,
now: Instant,
debounce: Duration,
reporter: &Option<WatchEventReporter>,
) {
if !event_kind_is_relevant(&event.kind) {
return;
}
let mut invalidated_repository_ids = Vec::new();
for path in event.paths {
let repository = {
let repositories_guard = repositories
.read()
.unwrap_or_else(|poisoned| poisoned.into_inner());
let repository_id = repository_id_for_path(
repositories_guard.values().cloned().collect::<Vec<_>>(),
&path,
);
repository_id.and_then(|repository_id| repositories_guard.get(&repository_id).cloned())
};
let Some(repository) = repository else {
continue;
};
if should_ignore_watch_path(&repository, &path) {
continue;
}
scheduler.record_path_change(&repository.repository_id, path.clone(), now, debounce);
validated_manifest_candidate_cache
.write()
.expect("validated manifest candidate cache poisoned")
.mark_dirty_root(&repository.root);
if !invalidated_repository_ids.contains(&repository.repository_id) {
if let Some(callback) = repository_cache_invalidation_callback {
callback(&repository.repository_id);
}
invalidated_repository_ids.push(repository.repository_id.clone());
}
info!(
repository_id = %repository.repository_id,
path = %path.display(),
"built-in watch mode accepted path change"
);
let relative_path =
repository_relative_watch_path(&repository, &path).unwrap_or_else(|| path.clone());
report_watch_event(
reporter,
WatchEvent::PathAccepted {
repository_id: repository.repository_id,
path: relative_path,
},
);
}
}
#[allow(clippy::too_many_arguments)]
fn handle_notify_dropped(
repositories: &Arc<RwLock<BTreeMap<String, WatchedRepository>>>,
scheduler: &mut WatchSchedulerState,
validated_manifest_candidate_cache: &Arc<RwLock<ValidatedManifestCandidateCache>>,
repository_cache_invalidation_callback: Option<&RepositoryCacheInvalidationCallback>,
error: String,
now: Instant,
debounce: Duration,
reporter: &Option<WatchEventReporter>,
) {
warn!(error = %error, "built-in watch mode dropped notify event");
report_watch_event(
reporter,
WatchEvent::NotifyDropped {
error: error.clone(),
},
);
let repositories = repositories
.read()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.values()
.cloned()
.collect::<Vec<_>>();
for repository in repositories {
scheduler.record_path_change(
&repository.repository_id,
repository.root.clone(),
now,
debounce,
);
validated_manifest_candidate_cache
.write()
.expect("validated manifest candidate cache poisoned")
.mark_dirty_root(&repository.root);
if let Some(callback) = repository_cache_invalidation_callback {
callback(&repository.repository_id);
}
info!(
repository_id = %repository.repository_id,
root = %repository.root.display(),
"built-in watch mode conservatively invalidated repository after dropped notify event"
);
}
}
fn invalidate_caches_for_stale_completion(
repository_id: &str,
_result: &Result<crate::indexer::IndexSummary, WatchRefreshFailure>,
repository_cache_invalidation_callback: Option<&RepositoryCacheInvalidationCallback>,
) -> bool {
if let Some(callback) = repository_cache_invalidation_callback {
callback(repository_id);
return true;
}
false
}
#[allow(clippy::too_many_arguments)]
fn handle_index_completed(
repositories: &Arc<RwLock<BTreeMap<String, WatchedRepository>>>,
scheduler: &mut WatchSchedulerState,
repository_id: &str,
class: WatchRefreshClass,
result: Result<crate::indexer::IndexSummary, WatchRefreshFailure>,
repository_cache_invalidation_callback: Option<&RepositoryCacheInvalidationCallback>,
now: Instant,
retry: Duration,
semantic_runtime: &SemanticRuntimeConfig,
semantic_credentials: &SemanticRuntimeCredentials,
reporter: &Option<WatchEventReporter>,
) {
let repository = repositories
.read()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.get(repository_id)
.cloned();
let Some(repository) = repository else {
return;
};
if let Some(callback) = repository_cache_invalidation_callback {
callback(&repository.repository_id);
}
match result {
Ok(summary) => {
scheduler.mark_succeeded(repository_id, class, now);
info!(
repository_id = %repository.repository_id,
root = %repository.root.display(),
refresh_class = %class.as_str(),
snapshot_id = %summary.snapshot_id,
files_scanned = summary.files_scanned,
files_changed = summary.files_changed,
files_deleted = summary.files_deleted,
duration_ms = summary.duration_ms,
"built-in watch mode refresh succeeded"
);
report_watch_event(
reporter,
WatchEvent::RefreshSucceeded {
repository_id: repository.repository_id.clone(),
root: repository.root.clone(),
refresh_class: class.as_str(),
summary: summary.clone(),
},
);
if class == WatchRefreshClass::ManifestFast {
queue_semantic_followup_if_needed(
&repository,
scheduler,
now,
semantic_runtime,
semantic_credentials,
reporter,
);
}
}
Err(failure) => {
if failure.blocks_retry() {
scheduler.mark_blocked(repository_id, class);
if reporter.is_none() {
warn!(
repository_id = %repository.repository_id,
root = %repository.root.display(),
refresh_class = %class.as_str(),
error = %failure.message,
"built-in watch mode refresh blocked; manual repair required"
);
}
report_watch_event(
reporter,
WatchEvent::RefreshBlocked {
repository_id: repository.repository_id,
root: repository.root,
refresh_class: class.as_str(),
reason: "storage_schema_incompatible",
error: failure.message,
},
);
} else {
let retry_delay = scheduler
.mark_failed(repository_id, class, now, retry)
.unwrap_or(retry);
if reporter.is_none() {
warn!(
repository_id = %repository.repository_id,
root = %repository.root.display(),
refresh_class = %class.as_str(),
retry_ms = retry_delay.as_millis(),
error = %failure.message,
"built-in watch mode refresh failed; retry scheduled"
);
}
report_watch_event(
reporter,
WatchEvent::RefreshFailed {
repository_id: repository.repository_id,
root: repository.root,
refresh_class: class.as_str(),
retry_ms: retry_delay.as_millis(),
error: failure.message,
},
);
}
}
}
}
fn queue_startup_refresh_if_needed(
repository: &WatchedRepository,
scheduler: &mut WatchSchedulerState,
now: Instant,
semantic_runtime: &SemanticRuntimeConfig,
semantic_credentials: &SemanticRuntimeCredentials,
debounce_ms: u64,
reporter: &Option<WatchEventReporter>,
) {
let startup_status = match startup_refresh_status(
repository,
semantic_runtime,
semantic_credentials,
) {
Ok(status) => status,
Err(error) => {
warn!(
repository_id = %repository.repository_id,
root = %repository.root.display(),
error = %error,
"built-in watch mode failed to evaluate startup freshness; queueing manifest-fast refresh"
);
scheduler.enqueue_initial_sync(
&repository.repository_id,
WatchRefreshClass::ManifestFast,
now,
);
report_watch_event(
reporter,
WatchEvent::RefreshQueued {
repository_id: repository.repository_id.clone(),
root: repository.root.clone(),
refresh_class: WatchRefreshClass::ManifestFast.as_str(),
reason: "startup_freshness_error".to_owned(),
snapshot_id: None,
debounce_ms: Some(debounce_ms),
},
);
return;
}
};
if !startup_status.should_refresh {
info!(
repository_id = %repository.repository_id,
root = %repository.root.display(),
snapshot_id = startup_status.snapshot_id.as_deref().unwrap_or("-"),
"built-in watch mode found refreshable startup state already satisfied"
);
report_watch_event(
reporter,
WatchEvent::StartupFresh {
repository_id: repository.repository_id.clone(),
root: repository.root.clone(),
snapshot_id: startup_status.snapshot_id,
},
);
return;
}
let class = startup_status
.refresh_class
.unwrap_or(WatchRefreshClass::ManifestFast);
scheduler.enqueue_initial_sync(&repository.repository_id, class, now);
info!(
repository_id = %repository.repository_id,
root = %repository.root.display(),
refresh_class = %class.as_str(),
startup_reason = %startup_status.reason,
snapshot_id = startup_status.snapshot_id.as_deref().unwrap_or("-"),
debounce_ms,
"built-in watch mode queued initial refresh"
);
report_watch_event(
reporter,
WatchEvent::RefreshQueued {
repository_id: repository.repository_id.clone(),
root: repository.root.clone(),
refresh_class: class.as_str(),
reason: startup_status.reason.to_owned(),
snapshot_id: startup_status.snapshot_id,
debounce_ms: Some(debounce_ms),
},
);
}
fn queue_semantic_followup_if_needed(
repository: &WatchedRepository,
scheduler: &mut WatchSchedulerState,
now: Instant,
semantic_runtime: &SemanticRuntimeConfig,
semantic_credentials: &SemanticRuntimeCredentials,
reporter: &Option<WatchEventReporter>,
) {
let status = match startup_refresh_status(repository, semantic_runtime, semantic_credentials) {
Ok(status) => status,
Err(error) => {
warn!(
repository_id = %repository.repository_id,
root = %repository.root.display(),
error = %error,
"built-in watch mode failed to evaluate semantic follow-up freshness; skipping"
);
return;
}
};
if status.refresh_class != Some(WatchRefreshClass::SemanticFollowup) {
return;
}
scheduler.enqueue_semantic_followup(&repository.repository_id, now);
info!(
repository_id = %repository.repository_id,
root = %repository.root.display(),
startup_reason = %status.reason,
snapshot_id = status.snapshot_id.as_deref().unwrap_or("-"),
"built-in watch mode queued semantic follow-up after manifest refresh"
);
report_watch_event(
reporter,
WatchEvent::RefreshQueued {
repository_id: repository.repository_id.clone(),
root: repository.root.clone(),
refresh_class: WatchRefreshClass::SemanticFollowup.as_str(),
reason: status.reason.to_owned(),
snapshot_id: status.snapshot_id,
debounce_ms: None,
},
);
}
#[cfg(test)]
mod tests {
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::time::Instant;
use super::*;
#[test]
fn repository_epochs_invalidate_stale_completion_across_detach_reattach() {
let mut epochs = RepositoryEpochs::default();
epochs.ensure("repo-001");
let dispatch_epoch = epochs.current("repo-001");
assert_eq!(dispatch_epoch, 0);
epochs.bump("repo-001");
epochs.ensure("repo-001");
assert_eq!(epochs.current("repo-001"), 1);
assert_ne!(dispatch_epoch, epochs.current("repo-001"));
let fresh_dispatch_epoch = epochs.current("repo-001");
assert_eq!(fresh_dispatch_epoch, epochs.current("repo-001"));
}
fn sample_index_summary() -> crate::indexer::IndexSummary {
crate::indexer::IndexSummary {
repository_id: "repo-001".to_owned(),
snapshot_id: "snap-1".to_owned(),
previous_snapshot_id: None,
snapshot_plan: "persist_new".to_owned(),
files_scanned: 1,
files_changed: 1,
files_deleted: 0,
changed_paths: vec!["src/lib.rs".to_owned()],
deleted_paths: Vec::new(),
semantic_refresh_mode: crate::indexer::SemanticRefreshMode::Disabled,
semantic_provider: None,
semantic_model: None,
semantic_records: 0,
diagnostics: Default::default(),
duration_ms: 1,
}
}
fn temp_supervisor_db_path(test_name: &str) -> PathBuf {
let unique = SystemTime::now()
.duration_since(UNIX_EPOCH)
.expect("system clock should be after unix epoch")
.as_nanos();
std::env::temp_dir().join(format!(
"frigg-watch-supervisor-{test_name}-{}-{unique}.sqlite3",
std::process::id()
))
}
fn seed_incompatible_schema_version(db_path: &std::path::Path) {
let conn =
rusqlite::Connection::open(db_path).expect("test incompatible sqlite db should open");
conn.execute_batch(
r#"
CREATE TABLE schema_version (
id INTEGER PRIMARY KEY CHECK (id = 1),
version INTEGER NOT NULL,
updated_at TEXT NOT NULL
);
INSERT INTO schema_version (id, version, updated_at)
VALUES (1, 10, CURRENT_TIMESTAMP);
"#,
)
.expect("test incompatible schema version should seed");
}
#[test]
fn stale_successful_completion_invalidates_repository_caches() {
let invalidated: Arc<std::sync::Mutex<Vec<String>>> =
Arc::new(std::sync::Mutex::new(Vec::new()));
let recorded = Arc::clone(&invalidated);
let callback: RepositoryCacheInvalidationCallback = Arc::new(move |repository_id: &str| {
recorded
.lock()
.expect("invalidation record poisoned")
.push(repository_id.to_owned());
});
let invoked = invalidate_caches_for_stale_completion(
"repo-001",
&Ok(sample_index_summary()),
Some(&callback),
);
assert!(invoked, "stale success must invalidate repository caches");
assert_eq!(
*invalidated.lock().expect("invalidation record poisoned"),
vec!["repo-001".to_owned()],
);
}
#[test]
fn stale_failed_completion_invalidates_repository_caches() {
let invalidated: Arc<std::sync::Mutex<Vec<String>>> =
Arc::new(std::sync::Mutex::new(Vec::new()));
let recorded = Arc::clone(&invalidated);
let callback: RepositoryCacheInvalidationCallback = Arc::new(move |repository_id: &str| {
recorded
.lock()
.expect("invalidation record poisoned")
.push(repository_id.to_owned());
});
let invoked = invalidate_caches_for_stale_completion(
"repo-001",
&Err(WatchRefreshFailure {
message: "boom".to_owned(),
kind: WatchRefreshFailureKind::Retryable,
}),
Some(&callback),
);
assert!(
invoked,
"stale failure must conservatively invalidate caches"
);
assert_eq!(
*invalidated.lock().expect("invalidation record poisoned"),
vec!["repo-001".to_owned()],
);
}
#[test]
fn repository_epochs_match_for_undisturbed_lifecycle() {
let mut epochs = RepositoryEpochs::default();
epochs.ensure("repo-001");
let dispatch_epoch = epochs.current("repo-001");
assert_eq!(dispatch_epoch, epochs.current("repo-001"));
}
#[test]
fn repository_epochs_default_to_zero_for_unknown_repository() {
let epochs = RepositoryEpochs::default();
assert_eq!(epochs.current("never-seen"), 0);
}
#[test]
fn active_changed_index_defers_manifest_fast_without_draining_scheduler() {
let now = Instant::now();
let retry = Duration::from_millis(250);
let mut scheduler = WatchSchedulerState::new(0);
scheduler.add_repository("repo-001");
scheduler.enqueue_initial_sync("repo-001", WatchRefreshClass::ManifestFast, now);
let mut registry = RuntimeTaskRegistry::new();
let task_id = registry.start_task(
RuntimeTaskKind::ChangedIndex,
"repo-001".to_owned(),
"watch_manifest_fast",
None,
);
let ready = scheduler
.next_ready_refresh(now)
.expect("manifest refresh should be ready");
assert_eq!(ready.class, WatchRefreshClass::ManifestFast);
assert!(active_refresh_task_running_for_repository(
®istry,
ready.class,
&ready.repository_id,
"stable-repo-001",
));
assert_eq!(
scheduler.mark_failed(&ready.repository_id, ready.class, now, retry),
Some(retry)
);
assert!(scheduler.repository_pending("repo-001", WatchRefreshClass::ManifestFast));
assert!(
scheduler.next_ready_refresh(now).is_none(),
"pending manifest refresh should back off while old task is active"
);
registry.finish_task(&task_id, RuntimeTaskStatus::Succeeded, None);
assert!(!active_refresh_task_running_for_repository(
®istry,
ready.class,
&ready.repository_id,
"stable-repo-001",
));
assert!(
scheduler.next_ready_refresh(now + retry).is_some(),
"manifest refresh should become ready after the active task finishes and retry elapses"
);
}
#[test]
fn refresh_task_dedup_uses_stable_and_runtime_aliases() {
let mut registry = RuntimeTaskRegistry::new();
let _task_id = registry.start_task(
RuntimeTaskKind::SemanticRefresh,
"stable-repo".to_owned(),
"semantic_attach_refresh",
None,
);
assert!(active_refresh_task_running_for_repository(
®istry,
WatchRefreshClass::SemanticFollowup,
"repo-001",
"stable-repo",
));
assert!(
!active_refresh_task_running_for_repository(
®istry,
WatchRefreshClass::ManifestFast,
"repo-001",
"stable-repo",
),
"semantic refresh should not block manifest-fast changed index"
);
}
#[test]
fn manifest_fast_defers_to_active_mcp_workspace_index() {
let mut registry = RuntimeTaskRegistry::new();
let _task_id = registry.start_task(
RuntimeTaskKind::WorkspaceIndex,
"stable-repo".to_owned(),
"workspace_index",
None,
);
assert!(
active_refresh_task_running_for_repository(
®istry,
WatchRefreshClass::ManifestFast,
"repo-001",
"stable-repo",
),
"manifest-fast must defer to an active MCP workspace_index on the same db"
);
assert!(
active_refresh_task_running_for_repository(
®istry,
WatchRefreshClass::SemanticFollowup,
"repo-001",
"stable-repo",
),
"semantic followup must also defer to an active MCP workspace_index"
);
}
#[test]
fn startup_refresh_eval_failure_queues_manifest_fast_instead_of_silent_drop() {
use super::super::repository::watched_repository_for_root;
use crate::settings::SemanticRuntimeProvider;
let now = Instant::now();
let repository = watched_repository_for_root(
"repo-001".to_owned(),
std::path::PathBuf::from("/nonexistent/frigg-watch-startup-eval"),
std::path::PathBuf::from("/nonexistent/frigg-watch-startup-eval/frigg.db"),
)
.expect("watched repository should build for a missing root");
let semantic_runtime = SemanticRuntimeConfig {
enabled: true,
provider: Some(SemanticRuntimeProvider::OpenAi),
model: Some("text-embedding-3-small".to_owned()),
strict_mode: false,
};
let credentials = SemanticRuntimeCredentials::default();
let mut scheduler = WatchSchedulerState::new(0);
scheduler.add_repository("repo-001");
queue_startup_refresh_if_needed(
&repository,
&mut scheduler,
now,
&semantic_runtime,
&credentials,
0,
&None,
);
assert!(
scheduler.repository_pending("repo-001", WatchRefreshClass::ManifestFast),
"a startup freshness evaluation failure must still queue a manifest-fast refresh"
);
}
#[test]
fn incompatible_storage_schema_errors_do_not_retry_forever() {
let failure = WatchRefreshFailure::from_error(FriggError::StorageSchemaIncompatible {
found_version: 10,
expected_version: 11,
db_path: "/tmp/frigg.db".into(),
});
assert_eq!(
failure.kind,
WatchRefreshFailureKind::StorageSchemaIncompatible
);
assert!(failure.blocks_retry());
assert!(failure.message.contains("storage schema is incompatible"));
let transient =
WatchRefreshFailure::from_error(FriggError::Internal("transient read failed".into()));
assert_eq!(transient.kind, WatchRefreshFailureKind::Retryable);
assert!(!transient.blocks_retry());
}
#[test]
fn incompatible_storage_schema_completion_blocks_watch_refresh() {
use super::super::repository::watched_repository_for_root;
let db_path = temp_supervisor_db_path("incompatible-completion");
seed_incompatible_schema_version(&db_path);
let storage_error = crate::storage::Storage::new(&db_path)
.require_current_schema()
.expect_err("incompatible schema should reject current-schema requirement");
let failure = WatchRefreshFailure::from_error(storage_error);
let root = std::env::temp_dir().join("frigg-watch-supervisor-incompatible-root");
let repository =
watched_repository_for_root("repo-001".to_owned(), root.clone(), db_path.clone())
.expect("watched repository should build for test root");
let repositories = Arc::new(RwLock::new(BTreeMap::from([(
repository.repository_id.clone(),
repository,
)])));
let mut scheduler = WatchSchedulerState::new(0);
scheduler.add_repository("repo-001");
let now = Instant::now();
let retry = Duration::from_millis(250);
scheduler.enqueue_initial_sync("repo-001", WatchRefreshClass::ManifestFast, now);
scheduler.mark_started("repo-001", WatchRefreshClass::ManifestFast);
let events = Arc::new(std::sync::Mutex::new(Vec::new()));
let recorded_events = Arc::clone(&events);
let reporter: WatchEventReporter = Arc::new(move |event| {
recorded_events
.lock()
.expect("watch event log poisoned")
.push(event);
});
handle_index_completed(
&repositories,
&mut scheduler,
"repo-001",
WatchRefreshClass::ManifestFast,
Err(failure),
None,
now,
retry,
&SemanticRuntimeConfig::default(),
&SemanticRuntimeCredentials::default(),
&Some(reporter),
);
assert!(!scheduler.repository_pending("repo-001", WatchRefreshClass::ManifestFast));
assert_eq!(scheduler.next_ready_refresh(now + retry), None);
let events = events.lock().expect("watch event log poisoned");
assert!(
events.iter().any(|event| matches!(
event,
WatchEvent::RefreshBlocked {
repository_id,
refresh_class,
reason,
error,
..
} if repository_id == "repo-001"
&& *refresh_class == WatchRefreshClass::ManifestFast.as_str()
&& *reason == "storage_schema_incompatible"
&& error.contains("storage schema is incompatible")
)),
"typed storage schema errors should emit RefreshBlocked"
);
assert!(
!events
.iter()
.any(|event| matches!(event, WatchEvent::RefreshFailed { .. })),
"typed storage schema errors must not schedule RefreshFailed retry events"
);
let _ = std::fs::remove_file(&db_path);
}
#[test]
fn manifest_fast_defers_to_active_mcp_workspace_prepare() {
let mut registry = RuntimeTaskRegistry::new();
let _task_id = registry.start_task(
RuntimeTaskKind::WorkspacePrepare,
"repo-001".to_owned(),
"workspace_prepare",
None,
);
assert!(
active_refresh_task_running_for_repository(
®istry,
WatchRefreshClass::ManifestFast,
"repo-001",
"repo-001",
),
"manifest-fast must defer to an active MCP workspace_prepare on the same db"
);
}
}