diff --git a/docs/session.md b/docs/session.md index 68b0f3b0..765bbbd9 100644 --- a/docs/session.md +++ b/docs/session.md @@ -597,11 +597,34 @@ pruning sessions whose source transcript disappeared, and backfilling repo identity for sessions indexed before it was resolvable. Deleting sessions that match `excluded_paths` still runs, restricted to the current scope. +### Scheduled Sync (macOS) + +```bash +recall sync schedule install --interval 30m +recall sync schedule status +recall sync schedule uninstall +``` + +The interval defaults to 30m; use an integer with `s`, `m`, or `h` units and a +minimum of 60 seconds. The LaunchAgent label is `io.github.samzong.recall.sync`; +its plist is `~/Library/LaunchAgents/io.github.samzong.recall.sync.plist` and +logs are `~/Library/Logs/recall/sync.stdout.log` and +`~/Library/Logs/recall/sync.stderr.log`. + +Install runs the first sync immediately. macOS may prompt for access to multiple +folders in sequence; click Allow within 60 seconds for each prompt. If the first +sync fails, rerun `recall sync schedule install` (include `--interval` again to +retain a custom interval). + +Git metadata probes time out after 60 seconds and abort the whole sync. The same +timeout applies to manual syncs and read commands that infer a project from the +current directory; read commands can bypass inference with `--project all`. +Linux and Windows report `not supported yet on `. + ## Backward Compatibility -- Existing commands keep working: `recall search`, `recall export`, - `recall import`, `recall usage`, `recall share init`, and the TUI remain - unchanged. +- Existing command names and flags remain available: `recall search`, + `recall export`, `recall import`, `recall usage`, `recall share init`, and the TUI. - Existing `recall export` remains the bulk export command. - Existing TUI shortcuts keep using the same internal session operations. - Export record schema is `v7`: event records retain `files` and nullable diff --git a/skills/recall/SKILL.md b/skills/recall/SKILL.md index 46a5fc7c..f0e42cd0 100644 --- a/skills/recall/SKILL.md +++ b/skills/recall/SKILL.md @@ -18,6 +18,14 @@ If neither MCP nor the installed CLI is available, stop and offer `brew install - `all` covers every indexed project. - The CLI derives an omitted `--project` from its working directory. Recall MCP treats an omitted `project` as global, so pass the current project unless the user explicitly requests all projects. +## Scheduled Sync + +- `recall sync schedule install --interval 30m` +- `recall sync schedule status` +- `recall sync schedule uninstall` + +Run install or uninstall only when the user explicitly requests a schedule change; run status only when the user asks to inspect it. + ## Find Sessions For a specific historical question, use MCP `search_messages` with keywords and the current project. Each match includes `session_id`, `seq`, `role`, and an excerpt around the match. Add `session_id` to search within a known session. This is full-text matching, not semantic search; use `search_sessions` for broader session discovery and `list_recent_sessions` when there is no query. diff --git a/src/cli.rs b/src/cli.rs index 848f01fe..661c14a4 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -50,8 +50,13 @@ enum Commands { #[command(subcommand)] command: session::SessionCommands, }, - #[command(about = "Scan configured AI coding session sources")] + #[command( + about = "Scan configured AI coding session sources", + args_conflicts_with_subcommands = true + )] Sync { + #[command(subcommand)] + command: Option, #[arg( long, conflicts_with_all = ["backfill_events", "dry_run", "force", "project"], @@ -310,6 +315,12 @@ pub(crate) fn run() -> Result<()> { match cli.command { Some(Commands::Info { format }) => crate::info::run(format)?, + Some(Commands::Sync { + command: Some(crate::schedule::SyncCommands::Schedule { command }), + .. + }) => { + crate::schedule::run(command)?; + } Some(Commands::Sync { session: Some(session), source, format, .. }) => { crate::sync::run_single_session(source.as_deref(), &session, format)?; } diff --git a/src/cli/tests.rs b/src/cli/tests.rs index dda40f61..e6bd9ac3 100644 --- a/src/cli/tests.rs +++ b/src/cli/tests.rs @@ -5,6 +5,64 @@ use super::{ use crate::session; use clap::{CommandFactory, Parser}; +#[test] +fn sync_schedule_preserves_manual_sync_and_excludes_mixed_arguments() { + use crate::schedule::{ScheduleCommands, SyncCommands}; + + let cli = Cli::try_parse_from(["recall", "sync", "schedule", "install"]).unwrap(); + assert!(matches!( + cli.command, + Some(Commands::Sync { + command: Some(SyncCommands::Schedule { + command: ScheduleCommands::Install { interval: 1800 } + }), + .. + }) + )); + let cli = + Cli::try_parse_from(["recall", "sync", "schedule", "install", "--interval", "2m"]).unwrap(); + assert!(matches!( + cli.command, + Some(Commands::Sync { + command: Some(SyncCommands::Schedule { + command: ScheduleCommands::Install { interval: 120 } + }), + .. + }) + )); + for action in ["status", "uninstall"] { + assert!(Cli::try_parse_from(["recall", "sync", "schedule", action]).is_ok()); + } + for args in [ + vec!["recall", "sync"], + vec!["recall", "sync", "--project", "all"], + vec!["recall", "sync", "--session", "native-id", "--source", "codex", "--format", "json"], + vec!["recall", "sync", "--source", "codex", "--force", "--verbose"], + vec!["recall", "sync", "--backfill-events", "--dry-run", "--project", "all"], + ] { + assert!(matches!( + Cli::try_parse_from(args).unwrap().command, + Some(Commands::Sync { command: None, .. }) + )); + } + for args in [ + vec!["recall", "sync", "--project", "all", "schedule", "install"], + vec![ + "recall", + "sync", + "--session", + "native-id", + "--source", + "codex", + "schedule", + "install", + ], + vec!["recall", "sync", "schedule", "status", "--force"], + ] { + assert!(Cli::try_parse_from(args).is_err()); + } +} + #[test] fn export_accepts_default_jsonl_without_format_flag() { let cli = Cli::try_parse_from([ diff --git a/src/db/project_store.rs b/src/db/project_store.rs index cb6ecd4b..5e48d68d 100644 --- a/src/db/project_store.rs +++ b/src/db/project_store.rs @@ -118,9 +118,10 @@ impl Store { scope: self.resolve_project_selector(project)?, inferred: false, }), - None => { - Ok(ScopeResolution { scope: crate::project_scope::auto_scope(), inferred: true }) - } + None => Ok(ScopeResolution { + scope: crate::project_scope::try_auto_scope()?, + inferred: true, + }), } } diff --git a/src/db/search/file_history.rs b/src/db/search/file_history.rs index 021c3ebf..0a813f6f 100644 --- a/src/db/search/file_history.rs +++ b/src/db/search/file_history.rs @@ -42,8 +42,8 @@ impl SearchEngine<'_> { let directory = directory .to_str() .ok_or_else(|| anyhow::anyhow!("invalid target directory"))?; - let root = git_toplevel(directory).unwrap_or_else(|| directory.to_string()); - let remote = origin_identity(&root).map(|identity| identity.remote); + let root = git_toplevel(directory)?.unwrap_or_else(|| directory.to_string()); + let remote = origin_identity(&root)?.map(|identity| identity.remote); (remote, Some(root)) } SelectorForm::Repository(identity) => (Some(identity.remote), None), @@ -73,7 +73,7 @@ impl SearchEngine<'_> { }; let mut absolute_path = None; let relative = if std::path::Path::new(path).is_absolute() || root.is_some() { - let resolved = RepoIdentityCache::default().resolve_file(path, root.as_deref()); + let resolved = RepoIdentityCache::default().resolve_file(path, root.as_deref())?; if let Some(file) = resolved.filter(|file| { remote.as_deref().is_some_and(|remote| file.repo_remote.as_deref() == Some(remote)) || root.as_deref().is_some_and(|root| { diff --git a/src/integration/regression/persistence.rs b/src/integration/regression/persistence.rs index 28b9b313..551cda0a 100644 --- a/src/integration/regression/persistence.rs +++ b/src/integration/regression/persistence.rs @@ -252,6 +252,7 @@ fn sync_resolves_cross_repository_files_and_keeps_native_paths() { if alternate.is_dir() { let resolved = crate::repo_identity::RepoIdentityCache::default() .resolve_file("src/deleted.rs", alternate.to_str()) + .unwrap() .unwrap(); assert_eq!(&resolved, file); } diff --git a/src/integration/roundtrip_conformance.rs b/src/integration/roundtrip_conformance.rs index 0d2ca11b..bf625a12 100644 --- a/src/integration/roundtrip_conformance.rs +++ b/src/integration/roundtrip_conformance.rs @@ -92,7 +92,7 @@ fn expected_contract(source: &str, raw: &RawSession) -> Value { .map(|event| { let mut files = event.files.clone(); for file in &mut files { - file.target = identities.resolve_file(&file.path, file.cwd.as_deref()); + file.target = identities.resolve_file(&file.path, file.cwd.as_deref()).unwrap(); } json!({ "event_seq": event.event_seq, diff --git a/src/lib.rs b/src/lib.rs index 7a622336..6481f66c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -19,6 +19,7 @@ pub(crate) mod project_scope; pub(crate) mod query; pub(crate) mod remote; pub(crate) mod repo_identity; +pub(crate) mod schedule; pub(crate) mod semantic; pub(crate) mod session; pub(crate) mod session_action; diff --git a/src/project_scope.rs b/src/project_scope.rs index a627c85c..203f2e31 100644 --- a/src/project_scope.rs +++ b/src/project_scope.rs @@ -145,29 +145,36 @@ pub(crate) fn classify_selector(value: &str) -> SelectorForm { } } -pub(crate) fn auto_scope_for_dir(dir: &Path) -> ProjectScope { +pub(crate) fn auto_scope_for_dir(dir: &Path) -> anyhow::Result { let Some(dir) = dir.to_str() else { - return ProjectScope::Global; + return Ok(ProjectScope::Global); }; - let Some(toplevel) = git_toplevel(dir) else { - return ProjectScope::Global; + let Some(toplevel) = git_toplevel(dir)? else { + return Ok(ProjectScope::Global); }; - match origin_identity(&toplevel) { + Ok(match origin_identity(&toplevel)? { Some(identity) => ProjectScope::Repository { filter: RepoFilter::Remote(identity.remote), local_root: Some(toplevel), }, None => ProjectScope::Directory(toplevel), - } + }) } -pub(crate) fn auto_scope() -> ProjectScope { +pub(crate) fn try_auto_scope() -> anyhow::Result { match std::env::current_dir() { Ok(dir) => auto_scope_for_dir(&dir), - Err(_) => ProjectScope::Global, + Err(_) => Ok(ProjectScope::Global), } } +pub(crate) fn auto_scope() -> ProjectScope { + try_auto_scope().unwrap_or_else(|error| { + eprintln!("Cannot infer project scope: {error}"); + ProjectScope::Global + }) +} + fn looks_like_path(value: &str) -> bool { value.starts_with('/') || value.starts_with('.') || value.starts_with('~') } @@ -235,7 +242,7 @@ mod tests { std::fs::create_dir_all(&nested).unwrap(); git_init(&root, Some("git@github.com:samzong/Recall.git")); - let scope = auto_scope_for_dir(&nested); + let scope = auto_scope_for_dir(&nested).unwrap(); let root_name = root.file_name().unwrap().to_str().unwrap().to_string(); match scope { @@ -254,7 +261,7 @@ mod tests { let root = temp_dir("recall-scope-noorigin"); git_init(&root, None); - let scope = auto_scope_for_dir(&root); + let scope = auto_scope_for_dir(&root).unwrap(); match scope { ProjectScope::Directory(directory) => { @@ -270,7 +277,7 @@ mod tests { fn auto_scope_is_global_outside_git() { let root = temp_dir("recall-scope-nongit"); - assert_eq!(auto_scope_for_dir(&root), ProjectScope::Global); + assert_eq!(auto_scope_for_dir(&root).unwrap(), ProjectScope::Global); std::fs::remove_dir_all(root).unwrap(); } diff --git a/src/repo_identity.rs b/src/repo_identity.rs index 1cdf8632..b08eeef7 100644 --- a/src/repo_identity.rs +++ b/src/repo_identity.rs @@ -1,6 +1,10 @@ use std::collections::HashMap; +use std::io::{Read, Seek}; use std::path::{Component, Path, PathBuf}; -use std::process::Command; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +use anyhow::{Context, Result}; use crate::types::FileTarget; @@ -19,72 +23,95 @@ pub(crate) struct RepoIdentityCache { } impl RepoIdentityCache { - pub(crate) fn resolve_file(&mut self, path: &str, cwd: Option<&str>) -> Option { + pub(crate) fn resolve_file( + &mut self, + path: &str, + cwd: Option<&str>, + ) -> Result> { if path.trim().is_empty() || path.contains('\0') || path.starts_with('~') { - return None; + return Ok(None); } if cfg!(not(windows)) && (path.starts_with("\\\\") || (path.as_bytes().get(1) == Some(&b':') && path.as_bytes().first().is_some_and(u8::is_ascii_alphabetic))) { - return None; + return Ok(None); } let path = Path::new(path); let absolute = if path.is_absolute() { path.to_path_buf() } else { if path.has_root() { - return None; + return Ok(None); } - let cwd = Path::new(cwd?); + let Some(cwd) = cwd else { return Ok(None) }; + let cwd = Path::new(cwd); if !cwd.is_absolute() { - return None; + return Ok(None); } cwd.join(path) }; - let absolute = normalize_file_path(&absolute)?; - let parent = absolute.parent()?.ancestors().find(|dir| dir.is_dir())?; - let root = self - .file_roots - .entry(parent.to_path_buf()) - .or_insert_with(|| git_toplevel(parent.to_str()?)) - .clone(); - let identity = root.as_deref().and_then(|root| self.resolve_toplevel(root)); + let Some(absolute) = normalize_file_path(&absolute) else { return Ok(None) }; + let Some(parent) = + absolute.parent().and_then(|parent| parent.ancestors().find(|dir| dir.is_dir())) + else { + return Ok(None); + }; + let root = match self.file_roots.get(parent) { + Some(root) => root.clone(), + None => { + let root = match parent.to_str() { + Some(parent) => git_toplevel(parent)?, + None => None, + }; + self.file_roots.insert(parent.to_path_buf(), root.clone()); + root + } + }; + let identity = + root.as_deref().map(|root| self.resolve_toplevel(root)).transpose()?.flatten(); let relative = root .as_deref() .and_then(|root| absolute.strip_prefix(root).ok()) .and_then(Path::to_str) .map(str::to_string); - Some(FileTarget { - absolute_path: absolute.to_str()?.to_string(), + let Some(absolute_path) = absolute.to_str() else { return Ok(None) }; + Ok(Some(FileTarget { + absolute_path: absolute_path.to_string(), repo_root: root, repo_relative_path: relative, repo_remote: identity.map(|repo| repo.remote), - }) + })) } - pub(crate) fn resolve(&mut self, directory: Option<&str>) -> Option { - let directory = directory?.trim(); + pub(crate) fn resolve(&mut self, directory: Option<&str>) -> Result> { + let Some(directory) = directory else { return Ok(None) }; + let directory = directory.trim(); if directory.is_empty() { - return None; + return Ok(None); } if let Some(identity) = self.by_directory.get(directory) { - return identity.clone(); + return Ok(identity.clone()); } - let Some(toplevel) = git_toplevel(directory) else { + let Some(toplevel) = git_toplevel(directory)? else { self.by_directory.insert(directory.to_string(), None); - return None; + return Ok(None); }; - let identity = self.resolve_toplevel(&toplevel); + let identity = self.resolve_toplevel(&toplevel)?; self.by_directory.insert(directory.to_string(), identity.clone()); - identity + Ok(identity) } - fn resolve_toplevel(&mut self, toplevel: &str) -> Option { - self.by_toplevel.entry(toplevel.into()).or_insert_with(|| origin_identity(toplevel)).clone() + fn resolve_toplevel(&mut self, toplevel: &str) -> Result> { + if let Some(identity) = self.by_toplevel.get(toplevel) { + return Ok(identity.clone()); + } + let identity = origin_identity(toplevel)?; + self.by_toplevel.insert(toplevel.to_string(), identity.clone()); + Ok(identity) } } @@ -105,13 +132,15 @@ fn normalize_file_path(path: &Path) -> Option { Some(result) } -pub(crate) fn git_toplevel(directory: &str) -> Option { +pub(crate) fn git_toplevel(directory: &str) -> Result> { git_output(["-C", directory, "rev-parse", "--show-toplevel"]) + .with_context(|| format!("cannot probe Git repository root for directory {directory:?}")) } -pub(crate) fn origin_identity(toplevel: &str) -> Option { - git_output(["-C", toplevel, "remote", "get-url", "origin"]) - .and_then(|url| normalize_remote_url(&url)) +pub(crate) fn origin_identity(toplevel: &str) -> Result> { + Ok(git_output(["-C", toplevel, "remote", "get-url", "origin"]) + .with_context(|| format!("cannot probe Git origin for directory {toplevel:?}"))? + .and_then(|url| normalize_remote_url(&url))) } /// Parses any Git remote form into a host-qualified identity: @@ -181,14 +210,69 @@ fn host_looks_like_host(host: &str) -> bool { host.contains('.') || host.eq_ignore_ascii_case("localhost") } -fn git_output(args: [&str; N]) -> Option { - let output = Command::new("git").args(args).output().ok()?; - if !output.status.success() { - return None; +pub(crate) const GIT_METADATA_PROBE_TIMEOUT: Duration = Duration::from_secs(60); + +fn git_output(args: [&str; N]) -> Result> { + probe_output(Command::new("git").args(args), GIT_METADATA_PROBE_TIMEOUT) +} + +fn stop_probe(child: &mut Child) -> Result<()> { + #[cfg(unix)] + { + let _ = Command::new("/bin/kill") + .args(["-KILL", "--", &format!("-{}", child.id())]) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status(); + } + let killed = child.kill(); + let reaped = child.wait(); + reaped.context("failed to reap Git metadata probe")?; + killed.context("failed to stop Git metadata probe") +} + +fn probe_output(command: &mut Command, timeout: Duration) -> Result> { + let mut output = tempfile::tempfile().context("cannot capture Git metadata probe output")?; + command.stdin(Stdio::null()).stdout(output.try_clone()?).stderr(Stdio::null()); + #[cfg(unix)] + { + use std::os::unix::process::CommandExt; + command.process_group(0); } - let value = String::from_utf8(output.stdout).ok()?; + let started = Instant::now(); + let mut child = match command.spawn() { + Ok(child) => child, + Err(_) => return Ok(None), + }; + let status = loop { + match child.try_wait() { + Ok(Some(status)) => break status, + Ok(None) if started.elapsed() < timeout => { + std::thread::sleep(Duration::from_millis(5)); + } + Ok(None) => { + stop_probe(&mut child)?; + anyhow::bail!( + "Git metadata probe timed out after {}s; it may be blocked by folder permissions or a stalled Git process. The operation was stopped; this is not evidence of index corruption. Check folder permissions, operating system privacy controls, and whether Git responds in the reported directory, then retry. Read commands can use `--project all` to bypass automatic project inference.", + timeout.as_secs() + ); + } + Err(error) => { + stop_probe(&mut child)?; + return Err(error).context("cannot wait for Git metadata probe"); + } + } + }; + if !status.success() { + return Ok(None); + } + output.rewind()?; + let mut bytes = Vec::new(); + output.read_to_end(&mut bytes)?; + let Ok(value) = String::from_utf8(bytes) else { return Ok(None) }; let value = value.trim(); - if value.is_empty() { None } else { Some(value.to_string()) } + Ok(if value.is_empty() { None } else { Some(value.to_string()) }) } #[cfg(test)] @@ -196,6 +280,30 @@ mod tests { use super::*; use std::fs; + #[cfg(unix)] + #[test] + fn timed_out_probe_stops_its_process_group_and_reaps_child() { + let temp = tempfile::tempdir().unwrap(); + let parent_pid = temp.path().join("parent.pid"); + let child_pid = temp.path().join("child.pid"); + let started = Instant::now(); + let result = probe_output( + Command::new("/bin/sh") + .args(["-c", "echo $$ > \"$1\"; sleep 30 & echo $! > \"$2\"; wait", "probe"]) + .arg(&parent_pid) + .arg(&child_pid), + Duration::from_millis(200), + ); + assert!(result.is_err()); + assert!(started.elapsed() < Duration::from_secs(3)); + for path in [parent_pid, child_pid] { + let pid = fs::read_to_string(path).unwrap(); + let output = + Command::new("/bin/ps").args(["-p", pid.trim(), "-o", "stat="]).output().unwrap(); + assert!(output.stdout.is_empty(), "probe process remains: {pid}"); + } + } + #[test] fn normalizes_github_remote_urls() { for url in [ @@ -253,7 +361,7 @@ mod tests { .unwrap(); let mut cache = RepoIdentityCache::default(); - let identity = cache.resolve(root.to_str()).unwrap(); + let identity = cache.resolve(root.to_str()).unwrap().unwrap(); assert_eq!(identity.remote, "github.com/samzong/Recall"); assert_eq!(identity.slug, "samzong/Recall"); assert_eq!(identity.name, "Recall"); @@ -266,7 +374,7 @@ mod tests { let root = std::env::temp_dir().join(format!("recall-non-git-{}", uuid::Uuid::new_v4())); fs::create_dir_all(&root).unwrap(); let mut cache = RepoIdentityCache::default(); - assert_eq!(cache.resolve(root.to_str()), None); + assert_eq!(cache.resolve(root.to_str()).unwrap(), None); fs::remove_dir_all(root).unwrap(); } @@ -279,6 +387,7 @@ mod tests { .unwrap(); let target = RepoIdentityCache::default() .resolve_file("link/../file.rs", temp.path().to_str()) + .unwrap() .unwrap(); assert_eq!( Path::new(&target.absolute_path), diff --git a/src/schedule.rs b/src/schedule.rs new file mode 100644 index 00000000..acbc761b --- /dev/null +++ b/src/schedule.rs @@ -0,0 +1,386 @@ +use anyhow::Result; +use clap::Subcommand; + +#[derive(Subcommand)] +pub(crate) enum SyncCommands { + #[command(about = "Manage scheduled global sync (macOS only)")] + Schedule { + #[command(subcommand)] + command: ScheduleCommands, + }, +} + +#[derive(Subcommand)] +pub(crate) enum ScheduleCommands { + #[command(about = "Install or reload a user LaunchAgent for recall sync --project all")] + Install { + #[arg(long, default_value = "30m", value_parser = parse_interval, help = "Sync interval: integer seconds (s), minutes (m), or hours (h); minimum 60s")] + interval: u32, + }, + #[command(about = "Unload and remove the scheduled sync LaunchAgent; retain logs")] + Uninstall, + #[command(about = "Show installation, load state, binary path, and recent log activity")] + Status, +} + +fn parse_interval(value: &str) -> std::result::Result { + let invalid = || { + "interval must be an integer followed by s, m, or h (for example 30m, 1h, or 900s), from 60s to 2147483647s".to_string() + }; + let (number, multiplier) = match value.as_bytes().last() { + Some(b's') => (&value[..value.len() - 1], 1), + Some(b'm') => (&value[..value.len() - 1], 60), + Some(b'h') => (&value[..value.len() - 1], 3600), + _ => return Err(invalid()), + }; + if number.is_empty() || !number.bytes().all(|byte| byte.is_ascii_digit()) { + return Err(invalid()); + } + number + .parse::() + .ok() + .and_then(|n| n.checked_mul(multiplier)) + .filter(|seconds| (60..=i32::MAX as u32).contains(seconds)) + .ok_or_else(invalid) +} + +pub(crate) fn run(command: ScheduleCommands) -> Result<()> { + #[cfg(target_os = "macos")] + { + macos::run(command) + } + #[cfg(not(target_os = "macos"))] + { + let _ = command; + unsupported(std::env::consts::OS) + } +} + +#[cfg(not(target_os = "macos"))] +fn unsupported(os: &str) -> Result<()> { + anyhow::bail!("not supported yet on {os}") +} + +#[cfg(target_os = "macos")] +mod macos { + use super::ScheduleCommands; + use anyhow::{Context, Result}; + use std::fs; + use std::io::Write; + use std::path::{Path, PathBuf}; + use std::process::{Command, Output}; + + const LABEL: &str = "io.github.samzong.recall.sync"; + + struct Agent { + plist: PathBuf, + stdout: PathBuf, + stderr: PathBuf, + domain: String, + } + + impl Agent { + fn new() -> Result { + let home = dirs::home_dir().context("cannot determine home directory")?; + let uid = checked(Command::new("/usr/bin/id").arg("-u"))?; + let uid = std::str::from_utf8(&uid.stdout)?.trim().parse::()?; + let logs = home.join("Library/Logs/recall"); + Ok(Self { + plist: home.join("Library/LaunchAgents").join(format!("{LABEL}.plist")), + stdout: logs.join("sync.stdout.log"), + stderr: logs.join("sync.stderr.log"), + domain: format!("gui/{uid}"), + }) + } + + fn target(&self) -> String { + format!("{}/{LABEL}", self.domain) + } + + fn loaded_job(&self) -> Result> { + let output = Command::new("/bin/launchctl").args(["print", &self.target()]).output()?; + if output.status.success() { + Ok(Some(String::from_utf8_lossy(&output.stdout).into_owned())) + } else if output.status.code() == Some(113) { + Ok(None) + } else { + command_error("launchctl print", &output) + } + } + + fn bootout(&self) -> Result<()> { + if self.loaded_job()?.is_some() { + checked(Command::new("/bin/launchctl").args(["bootout", &self.target()]))?; + } + Ok(()) + } + + fn install(&self, interval: u32) -> Result<()> { + let binary = std::env::current_exe().context("cannot resolve Recall executable")?; + let plist = render_plist(&binary, interval, &self.stdout, &self.stderr)?; + let directory = self.plist.parent().context("LaunchAgent path has no parent")?; + fs::create_dir_all(directory)?; + fs::create_dir_all(self.stdout.parent().context("log path has no parent")?)?; + let mut temporary = tempfile::NamedTempFile::new_in(directory)?; + temporary.write_all(plist.as_bytes())?; + checked(Command::new("/usr/bin/plutil").arg("-lint").arg(temporary.path()))?; + self.bootout()?; + temporary.persist(&self.plist).context("cannot write LaunchAgent plist")?; + checked(Command::new("/bin/launchctl").args(["bootstrap", &self.domain]).arg(&self.plist)) + .with_context(|| format!("could not load {}; inspect `recall sync schedule status` or remove it with `recall sync schedule uninstall`", self.plist.display()))?; + anyhow::ensure!( + self.loaded_job()?.is_some(), + "LaunchAgent was not loaded after bootstrap" + ); + checked(Command::new("/bin/launchctl").args(["kickstart", &self.target()])) + .context("could not start the first sync; inspect `recall sync schedule status` or remove it with `recall sync schedule uninstall`")?; + println!( + "Installed: {LABEL}\nInterval: {interval}s\nBinary: {}\nPlist: {}\nStdout: {}\nStderr: {}\nFirst sync: started now; macOS may ask for access to multiple folders in sequence (Documents, Downloads, Desktop, or other volumes). Click Allow within {} seconds for each prompt; if the first sync fails, rerun `recall sync schedule install`.", + binary.display(), + self.plist.display(), + self.stdout.display(), + self.stderr.display(), + crate::repo_identity::GIT_METADATA_PROBE_TIMEOUT.as_secs() + ); + Ok(()) + } + + fn uninstall(&self) -> Result<()> { + self.bootout()?; + match fs::remove_file(&self.plist) { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => return Err(error.into()), + } + println!( + "Not installed: {LABEL}\nLogs retained: {}", + self.stdout.parent().unwrap().display() + ); + Ok(()) + } + + fn status(&self) -> Result<()> { + let job = self.loaded_job()?; + let installed = self.plist.try_exists()?; + println!( + "{}: {LABEL}\nLoaded: {}\nPlist: {}", + if installed { "Installed" } else { "Not installed" }, + if job.is_some() { "yes" } else { "no" }, + self.plist.display() + ); + if let Some(job) = job { + println!("Last exit code: {}", last_exit_code(&job).unwrap_or("unavailable")); + } + if installed { + let output = checked( + Command::new("/usr/bin/plutil") + .args(["-convert", "json", "-o", "-"]) + .arg(&self.plist), + )?; + let definition: Definition = serde_json::from_slice(&output.stdout) + .context("cannot read scheduled sync configuration")?; + let binary = definition + .program_arguments + .first() + .context("LaunchAgent has no binary path")?; + println!( + "Interval: {}s\nBinary: {binary}\nBinary exists: {}", + definition.start_interval, + if Path::new(binary).try_exists()? { "yes" } else { "no" } + ); + log_status("Stdout", &definition.standard_out_path)?; + log_status("Stderr", &definition.standard_error_path)?; + } else { + log_status("Stdout", &self.stdout)?; + log_status("Stderr", &self.stderr)?; + } + Ok(()) + } + } + + fn last_exit_code(job: &str) -> Option<&str> { + job.lines().find_map(|line| line.trim().strip_prefix("last exit code = ")) + } + + #[derive(serde::Deserialize)] + #[serde(rename_all = "PascalCase")] + struct Definition { + program_arguments: Vec, + start_interval: u32, + standard_out_path: PathBuf, + standard_error_path: PathBuf, + } + + fn log_status(name: &str, path: &Path) -> Result<()> { + let modified = match fs::metadata(path) { + Ok(metadata) => { + chrono::DateTime::::from(metadata.modified()?).to_rfc3339() + } + Err(error) if error.kind() == std::io::ErrorKind::NotFound => "none".to_string(), + Err(error) => return Err(error.into()), + }; + println!("{name}: {}\n{name} last modified: {modified}", path.display()); + Ok(()) + } + + fn checked(command: &mut Command) -> Result { + let output = command + .output() + .with_context(|| format!("cannot run {}", command.get_program().to_string_lossy()))?; + if !output.status.success() { + return command_error(&command.get_program().to_string_lossy(), &output); + } + Ok(output) + } + + fn command_error(name: &str, output: &Output) -> Result { + anyhow::bail!( + "{name} exited with {}: {}", + output.status, + String::from_utf8_lossy(&output.stderr).trim() + ) + } + + fn xml_path(path: &Path) -> Result { + let text = path.to_str().context("LaunchAgent paths must be valid UTF-8")?; + anyhow::ensure!( + text.chars().all(|c| matches!(c, '\t' | '\n' | '\r') + || (c >= ' ' && c != '\u{fffe}' && c != '\u{ffff}')), + "LaunchAgent path contains a character unsupported by XML" + ); + Ok(text + .replace('&', "&") + .replace('<', "<") + .replace('>', ">") + .replace('\r', " ")) + } + + fn render_plist(binary: &Path, interval: u32, stdout: &Path, stderr: &Path) -> Result { + anyhow::ensure!(binary.is_absolute(), "Recall executable path must be absolute"); + let binary = xml_path(binary)?; + let stdout = xml_path(stdout)?; + let stderr = xml_path(stderr)?; + Ok(format!( + r#" + + + + Label{LABEL} + ProgramArguments + {binary}sync--projectall + StartInterval{interval} + KeepAlive + AbandonProcessGroup + StandardOutPath{stdout} + StandardErrorPath{stderr} + Umask63 + + +"# + )) + } + + pub(super) fn run(command: ScheduleCommands) -> Result<()> { + let agent = Agent::new()?; + match command { + ScheduleCommands::Install { interval } => agent.install(interval), + ScheduleCommands::Uninstall => agent.uninstall(), + ScheduleCommands::Status => agent.status(), + } + } + + #[cfg(test)] + mod tests { + use super::*; + + #[test] + fn launchctl_exit_status_preserves_codes_and_pending_state() { + for value in ["0", "7", "(never exited)"] { + let job = format!( + "gui/501/recall = {{\n\tstate = not running\n\tlast exit code = {value}\n}}" + ); + assert_eq!(last_exit_code(&job), Some(value)); + } + assert_eq!(last_exit_code("state = running\nlast terminating signal = 9"), None); + } + + #[test] + fn plist_roundtrips_paths_and_global_sync_arguments() { + let binary = Path::new("/Applications/Recall & /recall"); + let stdout = Path::new("/Users/example/Library/Logs/recall/out\r.log"); + let stderr = Path::new("/Users/example/Library/Logs/recall/err.log"); + let plist = render_plist(binary, 1800, stdout, stderr).unwrap(); + let file = tempfile::NamedTempFile::new().unwrap(); + fs::write(file.path(), plist).unwrap(); + let output = checked( + Command::new("/usr/bin/plutil") + .args(["-convert", "json", "-o", "-"]) + .arg(file.path()), + ) + .unwrap(); + let value: serde_json::Value = serde_json::from_slice(&output.stdout).unwrap(); + assert_eq!(value["Label"], LABEL); + assert_eq!( + value["ProgramArguments"], + serde_json::json!([binary.to_str().unwrap(), "sync", "--project", "all"]) + ); + assert_eq!(value["StartInterval"], 1800); + assert_eq!(value["KeepAlive"], false); + assert_eq!(value["AbandonProcessGroup"], true); + assert_eq!(value["Umask"], 0o77); + let definition: Definition = serde_json::from_slice(&output.stdout).unwrap(); + assert_eq!(definition.standard_out_path, stdout); + assert_eq!(definition.standard_error_path, stderr); + assert!(render_plist(Path::new("recall"), 1800, stdout, stderr).is_err()); + assert!(render_plist(Path::new("/bad\u{1}/recall"), 1800, stdout, stderr).is_err()); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn intervals_have_explicit_units_and_bounded_seconds() { + for (value, expected) in [ + ("30m", 1800), + ("1h", 3600), + ("900s", 900), + ("1m", 60), + ("2147483647s", i32::MAX as u32), + ] { + assert_eq!(parse_interval(value).unwrap(), expected); + } + for value in [ + "", + "30", + "s", + "0m", + "59s", + "1.5h", + "1h30m", + "-1m", + "+1m", + " 1m", + "2147483648s", + "4294967295h", + "999999999999s", + "时间", + ] { + assert!(parse_interval(value).is_err(), "accepted {value}"); + } + } + + #[cfg(not(target_os = "macos"))] + #[test] + fn all_schedule_actions_reject_the_current_platform() { + for command in [ + ScheduleCommands::Install { interval: 1800 }, + ScheduleCommands::Uninstall, + ScheduleCommands::Status, + ] { + assert!(run(command).is_err()); + } + } +} diff --git a/src/sync.rs b/src/sync.rs index d1edf10b..3284ae71 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -294,7 +294,7 @@ impl SyncJob { } let (mut existing, repo_backfill) = - self.resolve_existing_state(self.load_adapter_sync_context(source_id)?); + self.resolve_existing_state(self.load_adapter_sync_context(source_id)?)?; let settings = self.scan_settings(label); let mut streamed = 0; let mut sink_error = None; @@ -414,20 +414,20 @@ impl SyncJob { return Ok(()); } let context = self.load_adapter_sync_context(source)?; - let paths = context - .session_paths() - .filter(|path| { - path_matches_scope(&self.options.scope, path, &mut self.repo_cache) - && !self.path_excluder.as_ref().is_some_and(|matcher| { - paths_match_excluded( - path.directory.as_deref(), - path.source_file_path.as_deref(), - matcher, - ) - }) - }) - .cloned() - .collect::>(); + let mut paths = Vec::new(); + for path in context.session_paths() { + if path_matches_scope(&self.options.scope, path, &mut self.repo_cache)? + && !self.path_excluder.as_ref().is_some_and(|matcher| { + paths_match_excluded( + path.directory.as_deref(), + path.source_file_path.as_deref(), + matcher, + ) + }) + { + paths.push(path.clone()); + } + } let scan = self.scan_sessions(adapter, adapter.label(), &context)?; let Some(scan) = scan else { self.event_backfill.as_mut().unwrap().unavailable.push(source.to_string()); @@ -548,7 +548,7 @@ impl SyncJob { let Some(stored) = existing.paths.get_mut(&observation.source_id) else { continue; }; - if !path_matches_scope(&self.options.scope, stored, &mut self.repo_cache) { + if !path_matches_scope(&self.options.scope, stored, &mut self.repo_cache)? { continue; } if let Some(source_file_path) = observation.source_file_path.as_deref() @@ -666,7 +666,7 @@ impl SyncJob { source_id: &str, context: adapters::AdapterSyncContext, ) -> Result { - let (existing, repo_backfill) = self.resolve_existing_state(context); + let (existing, repo_backfill) = self.resolve_existing_state(context)?; self.write_repo_backfill(source_id, repo_backfill, &existing)?; Ok(existing) } @@ -693,7 +693,7 @@ impl SyncJob { fn resolve_existing_state( &mut self, context: adapters::AdapterSyncContext, - ) -> (ExistingState, Vec<(String, RepoIdentity)>) { + ) -> Result<(ExistingState, Vec<(String, RepoIdentity)>)> { let adapters::AdapterSyncContextParts { session_meta: meta, session_paths: mut paths, @@ -711,7 +711,7 @@ impl SyncJob { && (path.repo_remote.is_none() || path.repo_slug.is_none() || path.repo_name.is_none()) - && let Some(repo) = self.repo_cache.resolve(path.directory.as_deref()) + && let Some(repo) = self.repo_cache.resolve(path.directory.as_deref())? { path.repo_remote = Some(repo.remote.clone()); path.repo_slug = Some(repo.slug.clone()); @@ -721,7 +721,7 @@ impl SyncJob { } let existing = ExistingState { meta, paths, imported_ids, usage_meta, event_meta, metadata_meta }; - (existing, repo_backfill) + Ok((existing, repo_backfill)) } fn process_raw_session( @@ -736,9 +736,7 @@ impl SyncJob { return Ok(()); } - // Runs before every write and delete below, so a scoped sync can never - // touch a session outside its scope. - let repo_identity = self.repo_cache.resolve(raw.directory.as_deref()); + let repo_identity = self.repo_cache.resolve(raw.directory.as_deref())?; if !self .options .scope @@ -790,7 +788,7 @@ impl SyncJob { for event in &mut raw.events { for file in &mut event.files { - file.target = self.repo_cache.resolve_file(&file.path, file.cwd.as_deref()); + file.target = self.repo_cache.resolve_file(&file.path, file.cwd.as_deref())?; } } @@ -1414,9 +1412,9 @@ fn path_matches_scope( scope: &ProjectScope, path: &SessionPath, cache: &mut RepoIdentityCache, -) -> bool { - let repo = cache.resolve(path.directory.as_deref()); - scope.matches(SessionScopeFields { +) -> Result { + let repo = cache.resolve(path.directory.as_deref())?; + Ok(scope.matches(SessionScopeFields { directory: path.directory.as_deref(), repo_remote: path .repo_remote @@ -1430,7 +1428,7 @@ fn path_matches_scope( .repo_name .as_deref() .or_else(|| repo.as_ref().map(|repo| repo.name.as_str())), - }) + })) } fn paths_match_excluded( diff --git a/src/utils.rs b/src/utils.rs index ab840fee..16de39ea 100644 --- a/src/utils.rs +++ b/src/utils.rs @@ -69,7 +69,20 @@ pub(crate) fn try_acquire_sync_lock() -> anyhow::Result> { fn acquire_lock(path: &Path) -> anyhow::Result { let mut file = open_lock_file(path)?; - file.lock_exclusive()?; + match file.try_lock_exclusive() { + Ok(()) => {} + Err(error) if error.raw_os_error() == fs2::lock_contended_error().raw_os_error() => { + let owner = + std::fs::read_to_string(path).ok().and_then(|pid| pid.trim().parse::().ok()); + if let Some(pid) = owner { + eprintln!("Waiting for another Recall sync (pid {pid})"); + } else { + eprintln!("Waiting for another Recall sync"); + } + file.lock_exclusive()?; + } + Err(error) => return Err(error.into()), + } write_lock_owner(&mut file)?; Ok(file) } @@ -287,16 +300,6 @@ fn is_noise_first_message(content: &str) -> bool { mod tests { use super::*; - #[test] - fn lock_file_excludes_parallel_owner() { - let dir = tempfile::tempdir().unwrap(); - let path = dir.path().join("sync.lock"); - let first = acquire_lock(&path).unwrap(); - - assert!(try_acquire_lock(&path).unwrap().is_none()); - drop(first); - } - #[test] #[cfg(not(any(target_os = "macos", target_os = "windows")))] fn clipboard_candidates_fall_back_wl_copy_then_xclip_then_xsel_on_linux() {