From cad62857911b387a38b20e212c6a2e2bc323ad4e Mon Sep 17 00:00:00 2001 From: Junha Park <0xjunha@gmail.com> Date: Wed, 20 May 2026 20:07:44 +0900 Subject: [PATCH] feat(refresh): show session progress --- CHANGELOG.md | 2 + crates/cli/src/lib.rs | 1 + crates/cli/src/progress.rs | 458 +++++++++++++++++++++++++ crates/cli/src/refresh.rs | 139 ++++++-- crates/cli/src/share.rs | 441 ++++++------------------ crates/cli/src/tests/service_watch.rs | 69 +++- crates/cli/src/tests/share_progress.rs | 91 ++++- crates/core/src/index.rs | 24 +- crates/core/src/project/tests.rs | 14 + crates/core/src/project/types.rs | 10 + crates/core/src/project/workflow.rs | 43 ++- crates/core/src/sync.rs | 12 +- crates/index/src/engine.rs | 43 ++- crates/index/src/lib.rs | 3 +- crates/sync/src/engine.rs | 38 +- crates/sync/src/lib.rs | 4 +- 16 files changed, 1011 insertions(+), 381 deletions(-) create mode 100644 crates/cli/src/progress.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index b37e041..ffb60ff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,8 @@ All notable Darc release changes should be summarized here. ## Unreleased +- Show live refresh progress with spinner steps and session/project progress bars. + ## [0.2.1] - 2026-05-20 - Print post-upgrade auto-refresh restart guidance after `darc upgrade`. diff --git a/crates/cli/src/lib.rs b/crates/cli/src/lib.rs index a17f8a9..a364a65 100644 --- a/crates/cli/src/lib.rs +++ b/crates/cli/src/lib.rs @@ -4,6 +4,7 @@ compile_error!("Darc CLI does not support Windows."); mod agent_help; mod args; mod output; +mod progress; mod project; mod query_commands; mod refresh; diff --git a/crates/cli/src/progress.rs b/crates/cli/src/progress.rs new file mode 100644 index 0000000..d4cf38a --- /dev/null +++ b/crates/cli/src/progress.rs @@ -0,0 +1,458 @@ +use std::{ + io::{self, Write}, + sync::mpsc, + thread::{self, JoinHandle}, + time::{Duration, Instant}, +}; + +use crate::output::HumanStyle; + +const PROGRESS_BAR_WIDTH: usize = 24; +const PROGRESS_LABEL_WIDTH: usize = 18; +const PROGRESS_SPINNER_FRAMES: [&str; 10] = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]; +const PROGRESS_REDRAW_INTERVAL: Duration = Duration::from_millis(100); +const PROGRESS_COUNT_BUCKETS: u64 = 100; +const CLEAR_ACTIVE_LINE: &str = "\x1b[K"; + +/// Renders common CLI progress lines for interactive terminals. +pub(crate) struct ProgressOutput { + writer: W, + style: HumanStyle, + enabled: bool, + live_spinner: bool, + active_line: bool, + active_step: Option, + active_bar: Option, + step_index: usize, +} + +impl ProgressOutput { + /// Builds one common progress output from resolved terminal facts. + #[cfg(test)] + pub(crate) fn new(writer: W, style: HumanStyle, enabled: bool) -> Self { + Self::new_with_live_spinner(writer, style, enabled, false) + } + + /// Builds one common progress output with optional live step animation. + pub(crate) fn new_with_live_spinner( + writer: W, + style: HumanStyle, + enabled: bool, + live_spinner: bool, + ) -> Self { + Self { + writer, + style, + enabled, + live_spinner: enabled && live_spinner, + active_line: false, + active_step: None, + active_bar: None, + step_index: 0, + } + } + + /// Returns whether this output will render progress. + pub(crate) fn enabled(&self) -> bool { + self.enabled + } + + /// Returns the resolved human-output style. + pub(crate) fn style(&self) -> HumanStyle { + self.style + } + + /// Returns the configured progress writer. + pub(crate) fn writer_mut(&mut self) -> &mut W { + &mut self.writer + } + + /// Flushes the configured progress stream. + pub(crate) fn flush(&mut self) -> io::Result<()> { + self.writer.flush() + } + + /// Finishes any active progress row before the caller prints another message. + pub(crate) fn finish(&mut self) -> io::Result<()> { + if self.enabled { + self.finish_active_line()?; + self.flush()?; + } + Ok(()) + } + + /// Writes one operation heading and resets numbered steps. + pub(crate) fn heading(&mut self, message: &str) -> io::Result<()> { + self.finish_active_line()?; + self.step_index = 0; + writeln!(self.writer, "{message}") + } + + /// Writes one numbered step. + pub(crate) fn step(&mut self, message: &str) -> io::Result<()> { + self.step_with_indent("", message) + } + + /// Writes one numbered step beneath the given indentation. + pub(crate) fn step_with_indent(&mut self, indent: &str, message: &str) -> io::Result<()> { + self.finish_active_line()?; + self.step_index += 1; + if self.live_spinner { + let indent = indent.to_owned(); + let message = message.to_owned(); + write!( + self.writer, + "\r{}{}", + render_progress_step_line_with_indent( + self.style, + &indent, + self.step_index, + Some(PROGRESS_SPINNER_FRAMES[0]), + &message + ), + CLEAR_ACTIVE_LINE + )?; + self.writer.flush()?; + let spinner = LiveProgressStepSpinner::start( + self.style, + indent.clone(), + self.step_index, + message.clone(), + ); + self.active_step = Some(ActiveProgressStep { + indent, + index: self.step_index, + message, + spinner: Some(spinner), + }); + Ok(()) + } else { + writeln!( + self.writer, + "{}", + render_progress_step_line_with_indent( + self.style, + indent, + self.step_index, + None, + message + ) + ) + } + } + + /// Writes one in-place progress bar. + pub(crate) fn write_bar(&mut self, label: &str, current: u64, total: u64) -> io::Result<()> { + self.write_bar_with_indent("", label, current, total) + } + + /// Writes one in-place progress bar beneath the given indentation. + pub(crate) fn write_bar_with_indent( + &mut self, + indent: &str, + label: &str, + current: u64, + total: u64, + ) -> io::Result<()> { + self.write_bar_with_indent_at(indent, label, current, total, Instant::now()) + } + + /// Writes one in-place progress bar when the redraw budget allows it. + pub(crate) fn write_throttled_bar( + &mut self, + label: &str, + current: u64, + total: u64, + ) -> io::Result { + self.write_throttled_bar_with_indent("", label, current, total) + } + + /// Writes one indented progress bar when the redraw budget allows it. + pub(crate) fn write_throttled_bar_with_indent( + &mut self, + indent: &str, + label: &str, + current: u64, + total: u64, + ) -> io::Result { + if !self.should_render_bar(indent, label, current, total) { + return Ok(false); + } + let now = Instant::now(); + self.write_bar_with_indent_at(indent, label, current, total, now)?; + Ok(true) + } + + /// Writes one in-place percent progress bar when the redraw budget allows it. + pub(crate) fn write_throttled_percent_bar( + &mut self, + label: &str, + percent: u8, + ) -> io::Result { + if !self.should_render_bar("", label, u64::from(percent), 100) { + return Ok(false); + } + let now = Instant::now(); + self.write_percent_bar_at(label, percent, now)?; + Ok(true) + } + + /// Writes one in-place progress bar at a known render time. + fn write_bar_with_indent_at( + &mut self, + indent: &str, + label: &str, + current: u64, + total: u64, + rendered_at: Instant, + ) -> io::Result<()> { + self.finish_active_step()?; + let bar = render_progress_bar(current, total, PROGRESS_BAR_WIDTH, self.style); + let count = render_progress_count(current, total, self.style); + let percent = render_progress_percent(current, total, self.style); + write!( + self.writer, + "\r{indent} {label: io::Result<()> { + self.finish_active_step()?; + let bar = render_progress_bar(u64::from(percent), 100, PROGRESS_BAR_WIDTH, self.style); + let percent_text = render_percent(u64::from(percent), self.style); + write!( + self.writer, + "\r {label: io::Result<()> { + self.finish_active_step()?; + if self.active_line { + writeln!(self.writer)?; + self.active_line = false; + } + self.active_bar = None; + Ok(()) + } + + /// Finishes any active live step before rendering another progress shape. + fn finish_active_step(&mut self) -> io::Result<()> { + if let Some(mut step) = self.active_step.take() { + if let Some(spinner) = &mut step.spinner { + spinner.stop(); + } + writeln!( + self.writer, + "\r{}{}", + render_progress_step_line_with_indent( + self.style, + &step.indent, + step.index, + None, + &step.message + ), + CLEAR_ACTIVE_LINE + )?; + } + Ok(()) + } + + /// Returns whether the latest bar state should be written to the terminal. + fn should_render_bar(&mut self, indent: &str, label: &str, current: u64, total: u64) -> bool { + let Some(rendered) = &mut self.active_bar else { + return true; + }; + if rendered.indent != indent || rendered.label != label || rendered.total != total { + return true; + } + if rendered.current == current { + return false; + } + if current == 0 || current >= total { + return true; + } + if current < rendered.next_check_current { + return false; + } + rendered.next_check_current = next_progress_check_current(current, total); + Instant::now().duration_since(rendered.rendered_at) >= PROGRESS_REDRAW_INTERVAL + } +} + +/// Stores one active progress step currently animated by a spinner. +struct ActiveProgressStep { + indent: String, + index: usize, + message: String, + spinner: Option, +} + +/// Stores the last in-place progress bar rendered to the terminal. +struct ActiveProgressBar { + indent: String, + label: String, + current: u64, + total: u64, + rendered_at: Instant, + next_check_current: u64, +} + +/// Returns the next progress count that is worth checking against the time gate. +fn next_progress_check_current(current: u64, total: u64) -> u64 { + let stride = (total / PROGRESS_COUNT_BUCKETS).max(1); + current.saturating_add(stride).min(total) +} + +/// Animates one active progress step on stderr while blocking work runs. +struct LiveProgressStepSpinner { + stop: Option>, + handle: Option>, +} + +impl LiveProgressStepSpinner { + /// Starts one live progress step spinner on stderr. + fn start(style: HumanStyle, indent: String, step_index: usize, message: String) -> Self { + let (stop, stop_rx) = mpsc::channel(); + let handle = thread::spawn(move || { + let mut frame_index = 1; + let mut writer = io::stderr(); + while stop_rx.recv_timeout(Duration::from_millis(80)).is_err() { + let frame = PROGRESS_SPINNER_FRAMES[frame_index % PROGRESS_SPINNER_FRAMES.len()]; + let _ = write!( + writer, + "\r{}{}", + render_progress_step_line_with_indent( + style, + &indent, + step_index, + Some(frame), + &message + ), + CLEAR_ACTIVE_LINE + ); + let _ = writer.flush(); + frame_index += 1; + } + }); + Self { + stop: Some(stop), + handle: Some(handle), + } + } + + /// Stops the spinner thread and waits for it to exit. + fn stop(&mut self) { + if let Some(stop) = self.stop.take() { + let _ = stop.send(()); + } + if let Some(handle) = self.handle.take() { + let _ = handle.join(); + } + } +} + +impl Drop for LiveProgressStepSpinner { + /// Stops the spinner if its owner is dropped before normal completion. + fn drop(&mut self) { + self.stop(); + } +} + +/// Renders one numbered progress step with an optional spinner frame. +#[cfg(test)] +pub(crate) fn render_progress_step_line( + style: HumanStyle, + step_index: usize, + spinner: Option<&str>, + message: &str, +) -> String { + render_progress_step_line_with_indent(style, "", step_index, spinner, message) +} + +/// Renders one indented numbered progress step with an optional spinner frame. +fn render_progress_step_line_with_indent( + style: HumanStyle, + indent: &str, + step_index: usize, + spinner: Option<&str>, + message: &str, +) -> String { + let step = format!("[{}]", style.count(step_index)); + if let Some(spinner) = spinner { + format!("{indent} {} {step} {message}", style.path(spinner)) + } else { + format!("{indent} {step} {message}") + } +} + +/// Renders a fixed-width progress bar with a styled terminal variant. +fn render_progress_bar(current: u64, total: u64, width: usize, style: HumanStyle) -> String { + let filled = if total == 0 { + width + } else { + let bounded = current.min(total); + let width = u64::try_from(width).unwrap_or(u64::MAX); + let scaled = (u128::from(bounded) * u128::from(width)) / u128::from(total); + usize::try_from(scaled).unwrap_or(usize::MAX) + }; + let filled = filled.min(width); + let empty = width.saturating_sub(filled); + if style.enabled { + format!( + "{}{}", + style.ok("━".repeat(filled)), + style.muted("─".repeat(empty)) + ) + } else { + format!("[{}{}]", "#".repeat(filled), "-".repeat(empty)) + } +} + +/// Renders a fixed-width current/total progress count. +fn render_progress_count(current: u64, total: u64, style: HumanStyle) -> String { + let width = current.max(total).max(1).to_string().len(); + style.count(format!("{current:>width$}/{total}")) +} + +/// Renders the percentage for one current/total progress pair. +fn render_progress_percent(current: u64, total: u64, style: HumanStyle) -> String { + let percent = current + .min(total) + .saturating_mul(100) + .checked_div(total) + .unwrap_or(100); + render_percent(percent, style) +} + +/// Renders one right-aligned percentage. +fn render_percent(percent: u64, style: HumanStyle) -> String { + style.count(format!("{percent:>3}%")) +} diff --git a/crates/cli/src/refresh.rs b/crates/cli/src/refresh.rs index 6a3035c..5da49fb 100644 --- a/crates/cli/src/refresh.rs +++ b/crates/cli/src/refresh.rs @@ -23,6 +23,7 @@ use crate::output::{ HumanStyle, print_field, print_line, print_project_warning, print_section, stderr_progress_enabled, }; +use crate::progress::ProgressOutput; use crate::service::run_refresh_auto; use crate::sync_index::{ add_init_hint_for_unconfigured_project, format_skipped_rollout, format_sources, @@ -31,53 +32,69 @@ use crate::sync_index::{ /// Renders refresh progress events for interactive terminals. pub(crate) struct RefreshProgressPrinter { - pub(crate) writer: W, - pub(crate) style: HumanStyle, - pub(crate) enabled: bool, - pub(crate) total_projects: usize, + output: ProgressOutput, + total_projects: usize, + current_project_index: usize, } impl RefreshProgressPrinter { /// Builds one refresh progress printer for the current stderr stream. pub(crate) fn stderr() -> Self { - Self::new( + Self::new_with_live_spinner( io::stderr(), HumanStyle::stderr(), stderr_progress_enabled(), + true, ) } } impl RefreshProgressPrinter { /// Builds one refresh progress printer from resolved terminal facts. + #[cfg(test)] pub(crate) fn new(writer: W, style: HumanStyle, enabled: bool) -> Self { + Self::new_with_live_spinner(writer, style, enabled, false) + } + + /// Builds one refresh progress printer with optional live step animation. + fn new_with_live_spinner( + writer: W, + style: HumanStyle, + enabled: bool, + live_spinner: bool, + ) -> Self { Self { - writer, - style, - enabled, + output: ProgressOutput::new_with_live_spinner(writer, style, enabled, live_spinner), total_projects: 1, + current_project_index: 0, } } + /// Finishes any active progress row before the caller prints another message. + pub(crate) fn finish(&mut self) { + let _ = self.output.finish(); + } + /// Records one refresh progress event, ignoring presentation write failures. pub(crate) fn record(&mut self, event: RefreshProgress) { - if self.enabled { - let _ = self.write_event(event); - let _ = self.writer.flush(); + if self.output.enabled() && self.write_event(event).unwrap_or(false) { + let _ = self.output.flush(); } } /// Writes one refresh progress event to the configured stream. - pub(crate) fn write_event(&mut self, event: RefreshProgress) -> io::Result<()> { + pub(crate) fn write_event(&mut self, event: RefreshProgress) -> io::Result { match event { RefreshProgress::WorkspaceStarted { total_projects } => { self.total_projects = total_projects; - writeln!( - self.writer, + let style = self.output.style(); + let message = format!( "Refreshing workspace ({} project{})", - self.style.count(total_projects), + style.count(total_projects), if total_projects == 1 { "" } else { "s" } - ) + ); + self.output.heading(&message)?; + Ok(true) } RefreshProgress::ProjectStarted { project_name, @@ -86,38 +103,66 @@ impl RefreshProgressPrinter { total_projects, } => { self.total_projects = total_projects; + self.current_project_index = project_index; + let style = self.output.style(); if total_projects > 1 { - writeln!( - self.writer, + let message = format!( " [{}/{}] {}", - self.style.count(project_index), - self.style.count(total_projects), - self.style.bold(project_name) - ) + style.count(project_index), + style.count(total_projects), + style.bold(project_name) + ); + self.output.heading(&message)?; } else { - writeln!(self.writer, "Refreshing {}", self.style.bold(project_name)) + let message = format!("Refreshing {}", style.bold(project_name)); + self.output.heading(&message)?; } + Ok(true) } RefreshProgress::SyncStarted { project_name: _ } => { - writeln!(self.writer, "{}[1/2] Syncing archive...", self.indent()) + let indent = self.indent(); + self.output.step_with_indent(indent, "Syncing archive...")?; + Ok(true) } - RefreshProgress::SyncFinished { project_name: _ } => Ok(()), + RefreshProgress::SyncingSessions { + project_name: _, + synced_sessions, + total_sessions, + } => self.write_session_progress("Syncing sessions", synced_sessions, total_sessions), + RefreshProgress::SyncFinished { project_name: _ } => Ok(false), RefreshProgress::IndexStarted { project_name: _ } => { - writeln!(self.writer, "{}[2/2] Indexing sessions...", self.indent()) + let indent = self.indent(); + self.output + .step_with_indent(indent, "Indexing sessions...")?; + Ok(true) } - RefreshProgress::IndexFinished { project_name: _ } => Ok(()), + RefreshProgress::IndexingSessions { + project_name: _, + indexed_sessions, + total_sessions, + } => self.write_session_progress("Indexing sessions", indexed_sessions, total_sessions), + RefreshProgress::IndexFinished { project_name: _ } => Ok(false), RefreshProgress::ProjectFinished { project_name: _ } => { - writeln!(self.writer, "{}{}", self.indent(), self.style.ok("done"))?; - writeln!(self.writer) + self.output.finish_active_line()?; + let style = self.output.style(); + let indent = self.indent(); + writeln!(self.output.writer_mut(), "{}{}", indent, style.ok("done"))?; + self.write_workspace_project_bar()?; + writeln!(self.output.writer_mut())?; + Ok(true) } RefreshProgress::ProjectFailed { project_name: _ } => { + self.output.finish_active_line()?; + let style = self.output.style(); + let indent = self.indent(); writeln!( - self.writer, + self.output.writer_mut(), "{}{}", - self.indent(), - self.style.error("failed") + indent, + style.error("failed") )?; - writeln!(self.writer) + writeln!(self.output.writer_mut())?; + Ok(true) } } } @@ -130,6 +175,32 @@ impl RefreshProgressPrinter { " " } } + + /// Writes the workspace-level project bar after one project finishes. + fn write_workspace_project_bar(&mut self) -> io::Result<()> { + if self.total_projects <= 1 { + return Ok(()); + } + self.output.write_bar( + "Projects", + self.current_project_index as u64, + self.total_projects as u64, + )?; + self.output.finish_active_line() + } + + /// Writes a throttled session-count progress bar. + fn write_session_progress( + &mut self, + label: &'static str, + current: usize, + total: usize, + ) -> io::Result { + let current = current.min(total); + let indent = self.indent(); + self.output + .write_throttled_bar_with_indent(indent, label, current as u64, total as u64) + } } pub(crate) const DEFAULT_WATCH_DEBOUNCE: Duration = Duration::from_secs(30); @@ -362,6 +433,7 @@ pub(crate) fn run_refresh_once(request: &RefreshRunRequest) -> Result<()> { options, |event| progress.record(event), )?; + progress.finish(); print_refresh_all_report(&report); let result = refresh_all_exit_status(&report); return result; @@ -371,6 +443,7 @@ pub(crate) fn run_refresh_once(request: &RefreshRunRequest) -> Result<()> { progress.record(event); }) .map_err(add_init_hint_for_unconfigured_project)?; + progress.finish(); print_refresh_report(&report); Ok(()) } diff --git a/crates/cli/src/share.rs b/crates/cli/src/share.rs index 0a04410..b2d3bbb 100644 --- a/crates/cli/src/share.rs +++ b/crates/cli/src/share.rs @@ -1,12 +1,6 @@ use std::{ io::{self, Write}, path::PathBuf, - sync::{ - Arc, - atomic::{AtomicBool, Ordering}, - }, - thread::{self, JoinHandle}, - time::Duration, }; use anyhow::{Result, bail}; @@ -25,229 +19,14 @@ use crate::args::{ ShareRecipientCommands, ShareSessionSelectionArgs, }; use crate::output::{HumanStyle, stderr_progress_enabled}; +use crate::progress::ProgressOutput; +#[cfg(test)] +use crate::progress::render_progress_step_line; use crate::query_commands::provider_arg_to_source_kind; -const SHARE_PROGRESS_BAR_WIDTH: usize = 24; -const SHARE_PROGRESS_LABEL_WIDTH: usize = 18; -const SHARE_PROGRESS_SPINNER_FRAMES: [&str; 10] = - ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]; -const CLEAR_ACTIVE_LINE: &str = "\x1b[K"; - -/// Renders common share progress lines for interactive terminals. -struct ShareProgressOutput { - writer: W, - style: HumanStyle, - enabled: bool, - live_spinner: bool, - active_line: bool, - active_step: Option, - step_index: usize, -} - -impl ShareProgressOutput { - /// Builds one common share progress output from resolved terminal facts. - #[cfg(test)] - fn new(writer: W, style: HumanStyle, enabled: bool) -> Self { - Self::new_with_live_spinner(writer, style, enabled, false) - } - - /// Builds one common share progress output with optional live step animation. - fn new_with_live_spinner( - writer: W, - style: HumanStyle, - enabled: bool, - live_spinner: bool, - ) -> Self { - Self { - writer, - style, - enabled, - live_spinner: enabled && live_spinner, - active_line: false, - active_step: None, - step_index: 0, - } - } - - /// Returns whether this output will render progress. - fn enabled(&self) -> bool { - self.enabled - } - - /// Flushes the configured progress stream. - fn flush(&mut self) -> io::Result<()> { - self.writer.flush() - } - - /// Finishes any active progress row before the caller prints another message. - fn finish(&mut self) -> io::Result<()> { - if self.enabled { - self.finish_active_line()?; - self.flush()?; - } - Ok(()) - } - - /// Writes one operation heading and resets numbered steps. - fn heading(&mut self, message: &str) -> io::Result<()> { - self.finish_active_line()?; - self.step_index = 0; - writeln!(self.writer, "{message}") - } - - /// Writes one numbered step. - fn step(&mut self, message: &str) -> io::Result<()> { - self.finish_active_line()?; - self.step_index += 1; - if self.live_spinner { - let message = message.to_owned(); - write!( - self.writer, - "\r{}{}", - render_share_step_line( - self.style, - self.step_index, - Some(SHARE_PROGRESS_SPINNER_FRAMES[0]), - &message - ), - CLEAR_ACTIVE_LINE - )?; - self.writer.flush()?; - let spinner = LiveShareStepSpinner::start(self.style, self.step_index, message.clone()); - self.active_step = Some(ActiveShareStep { - index: self.step_index, - message, - spinner: Some(spinner), - }); - Ok(()) - } else { - writeln!( - self.writer, - "{}", - render_share_step_line(self.style, self.step_index, None, message) - ) - } - } - - /// Writes one in-place progress bar. - fn write_bar(&mut self, label: &str, current: u64, total: u64) -> io::Result<()> { - self.finish_active_step()?; - let bar = render_share_progress_bar(current, total, SHARE_PROGRESS_BAR_WIDTH, self.style); - let count = render_share_progress_count(current, total, self.style); - let percent = render_share_progress_percent(current, total, self.style); - write!( - self.writer, - "\r {label: io::Result<()> { - self.finish_active_step()?; - let bar = render_share_progress_bar( - u64::from(percent), - 100, - SHARE_PROGRESS_BAR_WIDTH, - self.style, - ); - let percent = render_share_percent(u64::from(percent), self.style); - write!( - self.writer, - "\r {label: io::Result<()> { - if let Some(mut step) = self.active_step.take() { - if let Some(spinner) = &mut step.spinner { - spinner.stop(); - } - writeln!( - self.writer, - "\r{}{}", - render_share_step_line(self.style, step.index, None, &step.message), - CLEAR_ACTIVE_LINE - )?; - } - Ok(()) - } - - /// Finishes any in-place progress line before writing regular output. - fn finish_active_line(&mut self) -> io::Result<()> { - self.finish_active_step()?; - if self.active_line { - writeln!(self.writer)?; - self.active_line = false; - } - Ok(()) - } -} - -/// Stores one active share step currently animated by a spinner. -struct ActiveShareStep { - index: usize, - message: String, - spinner: Option, -} - -/// Animates one active share step on stderr while blocking work runs. -struct LiveShareStepSpinner { - stop: Arc, - handle: Option>, -} - -impl LiveShareStepSpinner { - /// Starts one live share step spinner on stderr. - fn start(style: HumanStyle, step_index: usize, message: String) -> Self { - let stop = Arc::new(AtomicBool::new(false)); - let worker_stop = Arc::clone(&stop); - let handle = thread::spawn(move || { - let mut frame_index = 1; - let mut writer = io::stderr(); - while !worker_stop.load(Ordering::Relaxed) { - let frame = SHARE_PROGRESS_SPINNER_FRAMES - [frame_index % SHARE_PROGRESS_SPINNER_FRAMES.len()]; - let _ = write!( - writer, - "\r{}{}", - render_share_step_line(style, step_index, Some(frame), &message), - CLEAR_ACTIVE_LINE - ); - let _ = writer.flush(); - frame_index += 1; - thread::sleep(Duration::from_millis(80)); - } - }); - Self { - stop, - handle: Some(handle), - } - } - - /// Stops the spinner thread and waits for it to exit. - fn stop(&mut self) { - self.stop.store(true, Ordering::Relaxed); - if let Some(handle) = self.handle.take() { - let _ = handle.join(); - } - } -} - -impl Drop for LiveShareStepSpinner { - /// Stops the spinner if its owner is dropped before normal completion. - fn drop(&mut self) { - self.stop(); - } -} - /// Renders Darc share push progress for interactive terminals. pub(crate) struct SharePushProgressPrinter { - output: ShareProgressOutput, + output: ProgressOutput, rendering_session_progress: bool, } @@ -256,7 +35,7 @@ impl SharePushProgressPrinter { pub(crate) fn stderr() -> Self { let enabled = stderr_progress_enabled(); Self { - output: ShareProgressOutput::new_with_live_spinner( + output: ProgressOutput::new_with_live_spinner( io::stderr(), HumanStyle::stderr(), enabled, @@ -272,7 +51,7 @@ impl SharePushProgressPrinter { #[cfg(test)] pub(crate) fn new(writer: W, style: HumanStyle, enabled: bool) -> Self { Self { - output: ShareProgressOutput::new(writer, style, enabled), + output: ProgressOutput::new(writer, style, enabled), rendering_session_progress: false, } } @@ -289,62 +68,77 @@ impl SharePushProgressPrinter { /// Records one share push progress event, ignoring presentation write failures. pub(crate) fn record(&mut self, event: SharePushProgress) { - if self.output.enabled() { - let _ = self.write_event(event); + if self.output.enabled() && self.write_event(event).unwrap_or(false) { let _ = self.output.flush(); } } /// Writes one share push progress event to the configured stream. - pub(crate) fn write_event(&mut self, event: SharePushProgress) -> io::Result<()> { + pub(crate) fn write_event(&mut self, event: SharePushProgress) -> io::Result { match event { SharePushProgress::Started { git_branch, remote_name, remote_url, } => { + let style = self.output.style(); self.rendering_session_progress = false; let message = format!( "Pushing {} to {} ({})", - self.output.style.bold(git_branch), - self.output.style.bold(remote_name), + style.bold(git_branch), + style.bold(remote_name), remote_url ); - self.output.heading(&message) + self.output.heading(&message)?; + Ok(true) + } + SharePushProgress::PreparingCache => { + self.output.step("Preparing share cache...")?; + Ok(true) + } + SharePushProgress::FetchingRemote => { + self.output.step("Fetching remote branch...")?; + Ok(true) + } + SharePushProgress::HydratingLfs => { + self.output.step("Hydrating Git LFS objects...")?; + Ok(true) } - SharePushProgress::PreparingCache => self.output.step("Preparing share cache..."), - SharePushProgress::FetchingRemote => self.output.step("Fetching remote branch..."), - SharePushProgress::HydratingLfs => self.output.step("Hydrating Git LFS objects..."), SharePushProgress::ReadingCache => { - self.output.step("Reading cached share artifacts...") + self.output.step("Reading cached share artifacts...")?; + Ok(true) } SharePushProgress::ReusingPreviousExport { exported_turn_count, exported_session_count, } => { + let style = self.output.style(); let message = format!( "Reusing previous signed export ({} turns, {} sessions).", - self.output.style.count(exported_turn_count), - self.output.style.count(exported_session_count) + style.count(exported_turn_count), + style.count(exported_session_count) ); - self.output.step(&message) + self.output.step(&message)?; + Ok(true) } SharePushProgress::BuildingExport { total_turns } => { + let style = self.output.style(); let message = format!( "Building encrypted export ({} turns)...", - self.output.style.count(total_turns) + style.count(total_turns) ); - self.output.step(&message) + self.output.step(&message)?; + Ok(true) } SharePushProgress::ExportingTurns { exported_turns, total_turns, } => { if self.rendering_session_progress { - Ok(()) + Ok(false) } else { self.output - .write_bar("Exporting turns", exported_turns, total_turns) + .write_throttled_bar("Exporting turns", exported_turns, total_turns) } } SharePushProgress::ExportingSessions { @@ -352,54 +146,66 @@ impl SharePushProgressPrinter { total_sessions, } => { self.rendering_session_progress = true; - self.output - .write_bar("Exporting sessions", exported_sessions, total_sessions) + self.output.write_throttled_bar( + "Exporting sessions", + exported_sessions, + total_sessions, + ) } SharePushProgress::WritingMetadata { object_count } => { + let style = self.output.style(); let message = format!( "Writing share metadata ({} objects)...", - self.output.style.count(object_count) + style.count(object_count) ); - self.output.step(&message) + self.output.step(&message)?; + Ok(true) + } + SharePushProgress::Committing => { + self.output.step("Committing share artifacts...")?; + Ok(true) } - SharePushProgress::Committing => self.output.step("Committing share artifacts..."), SharePushProgress::Uploading { kind } => self.upload_step(kind), SharePushProgress::GitProgress { kind: _, message } => { self.write_git_progress(&message) } SharePushProgress::Finished { commit_id } => { self.output.finish_active_line()?; - let done = self.output.style.ok("done"); - let commit_id = self.output.style.muted(commit_id); - writeln!(self.output.writer, " {} {}", done, commit_id)?; - writeln!(self.output.writer) + let style = self.output.style(); + let done = style.ok("done"); + let commit_id = style.muted(commit_id); + writeln!(self.output.writer_mut(), " {} {}", done, commit_id)?; + writeln!(self.output.writer_mut()).map(|()| true) } - _ => Ok(()), + _ => Ok(false), } } /// Writes one upload phase step. - fn upload_step(&mut self, kind: ShareUploadKind) -> io::Result<()> { + fn upload_step(&mut self, kind: ShareUploadKind) -> io::Result { let message = match kind { ShareUploadKind::Lfs => "Uploading encrypted LFS objects...", ShareUploadKind::Git => "Uploading share branch...", _ => "Uploading share data...", }; - self.output.step(message) + self.output.step(message)?; + Ok(true) } /// Writes one streamed Git progress fragment. - fn write_git_progress(&mut self, message: &str) -> io::Result<()> { + fn write_git_progress(&mut self, message: &str) -> io::Result { if let Some(percent) = git_progress_percent(message) { - self.output.write_percent_bar("Uploading", percent)?; + return self + .output + .write_throttled_percent_bar("Uploading", percent); } - Ok(()) + Ok(false) } } /// Renders Darc share pull progress for interactive terminals. pub(crate) struct SharePullProgressPrinter { - output: ShareProgressOutput, + output: ProgressOutput, } impl SharePullProgressPrinter { @@ -407,7 +213,7 @@ impl SharePullProgressPrinter { pub(crate) fn stderr() -> Self { let enabled = stderr_progress_enabled(); Self { - output: ShareProgressOutput::new_with_live_spinner( + output: ProgressOutput::new_with_live_spinner( io::stderr(), HumanStyle::stderr(), enabled, @@ -422,7 +228,7 @@ impl SharePullProgressPrinter { #[cfg(test)] pub(crate) fn new(writer: W, style: HumanStyle, enabled: bool) -> Self { Self { - output: ShareProgressOutput::new(writer, style, enabled), + output: ProgressOutput::new(writer, style, enabled), } } @@ -438,119 +244,88 @@ impl SharePullProgressPrinter { /// Records one share pull progress event, ignoring presentation write failures. pub(crate) fn record(&mut self, event: SharePullProgress) { - if self.output.enabled() { - let _ = self.write_event(event); + if self.output.enabled() && self.write_event(event).unwrap_or(false) { let _ = self.output.flush(); } } /// Writes one share pull progress event to the configured stream. - pub(crate) fn write_event(&mut self, event: SharePullProgress) -> io::Result<()> { + pub(crate) fn write_event(&mut self, event: SharePullProgress) -> io::Result { match event { SharePullProgress::Started { git_branch, remote_name, remote_url, } => { + let style = self.output.style(); let message = format!( "Pulling {} from {} ({})", - self.output.style.bold(git_branch), - self.output.style.bold(remote_name), + style.bold(git_branch), + style.bold(remote_name), remote_url ); - self.output.heading(&message) + self.output.heading(&message)?; + Ok(true) + } + SharePullProgress::PreparingCache => { + self.output.step("Preparing share cache...")?; + Ok(true) + } + SharePullProgress::FetchingRemote => { + self.output.step("Fetching remote branch...")?; + Ok(true) + } + SharePullProgress::HydratingLfs => { + self.output.step("Hydrating Git LFS objects...")?; + Ok(true) } - SharePullProgress::PreparingCache => self.output.step("Preparing share cache..."), - SharePullProgress::FetchingRemote => self.output.step("Fetching remote branch..."), - SharePullProgress::HydratingLfs => self.output.step("Hydrating Git LFS objects..."), SharePullProgress::ReadingCache => { - self.output.step("Reading cached share artifacts...") + self.output.step("Reading cached share artifacts...")?; + Ok(true) } SharePullProgress::ImportingSessions { processed_sessions, total_sessions, - } => self - .output - .write_bar("Importing sessions", processed_sessions, total_sessions), + } => self.output.write_throttled_bar( + "Importing sessions", + processed_sessions, + total_sessions, + ), SharePullProgress::Finished { imported_turn_count, skipped_turn_count, warning_count, } => { self.output.finish_active_line()?; - let done = self.output.style.ok("done"); - let imported_turn_count = self.output.style.count(imported_turn_count); - let skipped_turn_count = self.output.style.count(skipped_turn_count); - let warning_count = self.output.style.count(warning_count); + let style = self.output.style(); + let done = style.ok("done"); + let imported_turn_count = style.count(imported_turn_count); + let skipped_turn_count = style.count(skipped_turn_count); + let warning_count = style.count(warning_count); writeln!( - self.output.writer, + self.output.writer_mut(), " {} imported {} turns, skipped {}, warnings {}", - done, imported_turn_count, skipped_turn_count, warning_count + done, + imported_turn_count, + skipped_turn_count, + warning_count )?; - writeln!(self.output.writer) + writeln!(self.output.writer_mut()).map(|()| true) } - _ => Ok(()), + _ => Ok(false), } } } /// Renders one numbered share step with an optional spinner frame. +#[cfg(test)] pub(crate) fn render_share_step_line( style: HumanStyle, step_index: usize, spinner: Option<&str>, message: &str, ) -> String { - let step = format!("[{}]", style.count(step_index)); - if let Some(spinner) = spinner { - format!(" {} {step} {message}", style.path(spinner)) - } else { - format!(" {step} {message}") - } -} - -/// Renders a fixed-width progress bar with a styled terminal variant. -fn render_share_progress_bar(current: u64, total: u64, width: usize, style: HumanStyle) -> String { - let filled = if total == 0 { - width - } else { - let bounded = current.min(total); - let width = u64::try_from(width).unwrap_or(u64::MAX); - let scaled = (u128::from(bounded) * u128::from(width)) / u128::from(total); - usize::try_from(scaled).unwrap_or(usize::MAX) - }; - let filled = filled.min(width); - let empty = width.saturating_sub(filled); - if style.enabled { - format!( - "{}{}", - style.ok("━".repeat(filled)), - style.muted("─".repeat(empty)) - ) - } else { - format!("[{}{}]", "#".repeat(filled), "-".repeat(empty)) - } -} - -/// Renders a fixed-width current/total progress count. -fn render_share_progress_count(current: u64, total: u64, style: HumanStyle) -> String { - let width = current.max(total).max(1).to_string().len(); - style.count(format!("{current:>width$}/{total}")) -} - -/// Renders the percentage for one current/total progress pair. -fn render_share_progress_percent(current: u64, total: u64, style: HumanStyle) -> String { - let percent = current - .min(total) - .saturating_mul(100) - .checked_div(total) - .unwrap_or(100); - render_share_percent(percent, style) -} - -/// Renders one right-aligned percentage. -fn render_share_percent(percent: u64, style: HumanStyle) -> String { - style.count(format!("{percent:>3}%")) + render_progress_step_line(style, step_index, spinner, message) } /// Extracts the last integer percentage from one Git progress fragment. diff --git a/crates/cli/src/tests/service_watch.rs b/crates/cli/src/tests/service_watch.rs index 3c28387..08d2931 100644 --- a/crates/cli/src/tests/service_watch.rs +++ b/crates/cli/src/tests/service_watch.rs @@ -1,5 +1,25 @@ +use std::io::{self, Write}; + use super::*; +#[derive(Default)] +struct FlushCountingWriter { + bytes: Vec, + flushes: usize, +} + +impl Write for FlushCountingWriter { + fn write(&mut self, buf: &[u8]) -> io::Result { + self.bytes.extend_from_slice(buf); + Ok(buf.len()) + } + + fn flush(&mut self) -> io::Result<()> { + self.flushes += 1; + Ok(()) + } +} + #[test] fn refresh_command_accepts_provider_filters_and_all() { let cli = Cli::try_parse_from(["darc", "refresh", "--provider", "claude", "--all"]).unwrap(); @@ -975,9 +995,25 @@ fn refresh_progress_printer_writes_interactive_steps() { printer.record(RefreshProgress::SyncStarted { project_name: "repo-a".to_owned(), }); + printer.record(RefreshProgress::SyncingSessions { + project_name: "repo-a".to_owned(), + synced_sessions: 2, + total_sessions: 4, + }); + printer.record(RefreshProgress::SyncFinished { + project_name: "repo-a".to_owned(), + }); printer.record(RefreshProgress::IndexStarted { project_name: "repo-a".to_owned(), }); + printer.record(RefreshProgress::IndexingSessions { + project_name: "repo-a".to_owned(), + indexed_sessions: 3, + total_sessions: 4, + }); + printer.record(RefreshProgress::IndexFinished { + project_name: "repo-a".to_owned(), + }); printer.record(RefreshProgress::ProjectFinished { project_name: "repo-a".to_owned(), }); @@ -986,9 +1022,12 @@ fn refresh_progress_printer_writes_interactive_steps() { let output = String::from_utf8(output).unwrap(); assert!(output.contains("Refreshing workspace (2 projects)")); assert!(output.contains(" [1/2] repo-a")); - assert!(output.contains(" [1/2] Syncing archive...")); - assert!(output.contains(" [2/2] Indexing sessions...")); + assert!(output.contains(" [1] Syncing archive...")); + assert!(output.contains(" Syncing sessions [############------------] 2/4 50%")); + assert!(output.contains(" [2] Indexing sessions...")); + assert!(output.contains(" Indexing sessions [##################------] 3/4 75%")); assert!(output.contains(" done")); + assert!(output.contains("Projects [############------------] 1/2 50%")); } #[test] @@ -1011,6 +1050,32 @@ fn refresh_progress_printer_stays_silent_when_disabled() { assert!(output.is_empty()); } +#[test] +fn refresh_progress_printer_does_not_flush_skipped_session_updates() { + let mut output = FlushCountingWriter::default(); + { + let style = super::HumanStyle::new(false, false, None); + let mut printer = super::RefreshProgressPrinter::new(&mut output, style, true); + printer.record(RefreshProgress::IndexStarted { + project_name: "repo-a".to_owned(), + }); + printer.record(RefreshProgress::IndexingSessions { + project_name: "repo-a".to_owned(), + indexed_sessions: 0, + total_sessions: 1_000, + }); + for indexed_sessions in 1..100 { + printer.record(RefreshProgress::IndexingSessions { + project_name: "repo-a".to_owned(), + indexed_sessions, + total_sessions: 1_000, + }); + } + } + + assert_eq!(output.flushes, 2); +} + #[test] fn service_progress_printer_writes_interactive_steps() { let mut output = Vec::new(); diff --git a/crates/cli/src/tests/share_progress.rs b/crates/cli/src/tests/share_progress.rs index 2bd37b8..2386ac7 100644 --- a/crates/cli/src/tests/share_progress.rs +++ b/crates/cli/src/tests/share_progress.rs @@ -1,4 +1,7 @@ -use std::{cell::Cell, io::Write}; +use std::{ + cell::Cell, + io::{self, Write}, +}; use darc_core::{ ShareFetchReport, ShareMergeReport, SharePullProgress, SharePullReport, SharePushProgress, @@ -7,6 +10,24 @@ use darc_core::{ use super::*; +#[derive(Default)] +struct FlushCountingWriter { + bytes: Vec, + flushes: usize, +} + +impl Write for FlushCountingWriter { + fn write(&mut self, buf: &[u8]) -> io::Result { + self.bytes.extend_from_slice(buf); + Ok(buf.len()) + } + + fn flush(&mut self) -> io::Result<()> { + self.flushes += 1; + Ok(()) + } +} + fn sample_share_push_report() -> SharePushReport { SharePushReport { branch: "team".to_owned(), @@ -44,6 +65,74 @@ fn sample_share_pull_report() -> SharePullReport { } } +#[test] +fn share_push_progress_printer_throttles_hot_session_updates() { + let mut output = FlushCountingWriter::default(); + { + let style = super::HumanStyle::new(false, false, None); + let mut printer = super::share::SharePushProgressPrinter::new(&mut output, style, true); + printer.record(SharePushProgress::ExportingSessions { + exported_sessions: 0, + total_sessions: 10_000, + }); + for exported_sessions in 1..10_000 { + printer.record(SharePushProgress::ExportingSessions { + exported_sessions, + total_sessions: 10_000, + }); + } + printer.record(SharePushProgress::ExportingSessions { + exported_sessions: 10_000, + total_sessions: 10_000, + }); + } + + assert_eq!(output.flushes, 2); +} + +#[test] +fn share_pull_progress_printer_throttles_hot_session_updates() { + let mut output = FlushCountingWriter::default(); + { + let style = super::HumanStyle::new(false, false, None); + let mut printer = super::share::SharePullProgressPrinter::new(&mut output, style, true); + printer.record(SharePullProgress::ImportingSessions { + processed_sessions: 0, + total_sessions: 10_000, + }); + for processed_sessions in 1..10_000 { + printer.record(SharePullProgress::ImportingSessions { + processed_sessions, + total_sessions: 10_000, + }); + } + printer.record(SharePullProgress::ImportingSessions { + processed_sessions: 10_000, + total_sessions: 10_000, + }); + } + + assert_eq!(output.flushes, 2); +} + +#[test] +fn share_push_progress_printer_ignores_unrendered_git_fragments_without_flush() { + let mut output = FlushCountingWriter::default(); + { + let style = super::HumanStyle::new(false, false, None); + let mut printer = super::share::SharePushProgressPrinter::new(&mut output, style, true); + for _ in 0..100 { + printer.record(SharePushProgress::GitProgress { + kind: ShareUploadKind::Git, + message: "Counting objects: synthetic diagnostic".to_owned(), + }); + } + } + + assert_eq!(output.flushes, 0); + assert!(output.bytes.is_empty()); +} + #[test] fn share_push_progress_printer_writes_session_and_upload_bars() { let mut output = Vec::new(); diff --git a/crates/core/src/index.rs b/crates/core/src/index.rs index 189549e..0ddd081 100644 --- a/crates/core/src/index.rs +++ b/crates/core/src/index.rs @@ -7,8 +7,12 @@ use std::{ }; use anyhow::{Context, Result}; +pub(crate) use darc_index::IndexProgress; pub use darc_index::{IndexReport, SkippedCodexRollout, SkippedRollout}; -use darc_index::{ProjectIndexRequest, index_project_archived_sessions}; +use darc_index::{ + ProjectIndexRequest, index_project_archived_sessions, + index_project_archived_sessions_with_progress, +}; use darc_paths::SourceKind; use darc_store::{ INDEX_DB_FILE_NAME, preserve_index_sharing_state_for_projects, remove_index_database, @@ -182,6 +186,22 @@ pub(crate) fn index_project_sessions_for_active_project( active_project: ActiveProject, root: PathBuf, providers: &[SourceKind], +) -> Result { + let mut progress = |_| {}; + index_project_sessions_for_active_project_with_progress( + active_project, + root, + providers, + &mut progress, + ) +} + +/// Indexes archived provider rollouts while reporting session progress. +pub(crate) fn index_project_sessions_for_active_project_with_progress( + active_project: ActiveProject, + root: PathBuf, + providers: &[SourceKind], + progress: impl FnMut(IndexProgress), ) -> Result { let request = ProjectIndexRequest { project_id: active_project.project.id, @@ -190,7 +210,7 @@ pub(crate) fn index_project_sessions_for_active_project( sessions_root: active_project.project.sessions_root, index_db_path: root.join(INDEX_DB_FILE_NAME), }; - index_project_archived_sessions(&request, providers) + index_project_archived_sessions_with_progress(&request, providers, progress) } /// Resolves the selected provider list for one indexing run. diff --git a/crates/core/src/project/tests.rs b/crates/core/src/project/tests.rs index 7c70409..84b8676 100644 --- a/crates/core/src/project/tests.rs +++ b/crates/core/src/project/tests.rs @@ -899,12 +899,22 @@ fn refresh_all_projects_best_effort_reports_progress_events() -> Result<()> { RefreshProgress::SyncStarted { project_name } => { format!("sync-start:{project_name}") } + RefreshProgress::SyncingSessions { + project_name, + synced_sessions, + total_sessions, + } => format!("sync-sessions:{project_name}:{synced_sessions}/{total_sessions}"), RefreshProgress::SyncFinished { project_name } => { format!("sync-finish:{project_name}") } RefreshProgress::IndexStarted { project_name } => { format!("index-start:{project_name}") } + RefreshProgress::IndexingSessions { + project_name, + indexed_sessions, + total_sessions, + } => format!("index-sessions:{project_name}:{indexed_sessions}/{total_sessions}"), RefreshProgress::IndexFinished { project_name } => { format!("index-finish:{project_name}") } @@ -928,8 +938,12 @@ fn refresh_all_projects_best_effort_reports_progress_events() -> Result<()> { "project-failed:broken-repo", "project-start:2/2:healthy-repo", "sync-start:healthy-repo", + "sync-sessions:healthy-repo:0/1", + "sync-sessions:healthy-repo:1/1", "sync-finish:healthy-repo", "index-start:healthy-repo", + "index-sessions:healthy-repo:0/1", + "index-sessions:healthy-repo:1/1", "index-finish:healthy-repo", "project-finish:healthy-repo", ] diff --git a/crates/core/src/project/types.rs b/crates/core/src/project/types.rs index 57c06ea..8138cb4 100644 --- a/crates/core/src/project/types.rs +++ b/crates/core/src/project/types.rs @@ -69,12 +69,22 @@ pub enum RefreshProgress { SyncStarted { project_name: String, }, + SyncingSessions { + project_name: String, + synced_sessions: usize, + total_sessions: usize, + }, SyncFinished { project_name: String, }, IndexStarted { project_name: String, }, + IndexingSessions { + project_name: String, + indexed_sessions: usize, + total_sessions: usize, + }, IndexFinished { project_name: String, }, diff --git a/crates/core/src/project/workflow.rs b/crates/core/src/project/workflow.rs index 2d0693c..f6bec94 100644 --- a/crates/core/src/project/workflow.rs +++ b/crates/core/src/project/workflow.rs @@ -22,8 +22,13 @@ use crate::{ active_project::{ActiveProject, load_active_project}, config::ProjectConfig, default_root_path, - index::{index_project_sessions_for_active_project, selected_index_providers}, - sync::{SyncOptions, execute_sync, prepare_sync_for_active_project}, + index::{ + IndexProgress, index_project_sessions_for_active_project_with_progress, + selected_index_providers, + }, + sync::{ + SyncOptions, SyncProgress, execute_sync_with_progress, prepare_sync_for_active_project, + }, }; /// Stores display identity for one project refresh progress entry. @@ -407,12 +412,29 @@ fn refresh_loaded_project_from( progress(RefreshProgress::SyncStarted { project_name: project_name.to_owned(), }); - let sync = execute_sync(prepare_sync_for_active_project( + let sync_plan = prepare_sync_for_active_project( active_project, SyncOptions { provider_filter: options.provider_filter.clone(), }, - )?)?; + )?; + let sync_unchanged_sessions = sync_plan.sessions_unchanged; + let sync_total_sessions = sync_unchanged_sessions + sync_plan.sessions_to_copy(); + progress(RefreshProgress::SyncingSessions { + project_name: project_name.to_owned(), + synced_sessions: sync_unchanged_sessions, + total_sessions: sync_total_sessions, + }); + let sync = execute_sync_with_progress(sync_plan, |event| { + let SyncProgress::CopyingSessions { + copied_sessions, .. + } = event; + progress(RefreshProgress::SyncingSessions { + project_name: project_name.to_owned(), + synced_sessions: sync_unchanged_sessions + copied_sessions, + total_sessions: sync_total_sessions, + }); + })?; progress(RefreshProgress::SyncFinished { project_name: project_name.to_owned(), }); @@ -420,10 +442,21 @@ fn refresh_loaded_project_from( progress(RefreshProgress::IndexStarted { project_name: project_name.to_owned(), }); - let index = index_project_sessions_for_active_project( + let index = index_project_sessions_for_active_project_with_progress( index_project, root, &selected_index_providers(&options.provider_filter), + |event| { + let IndexProgress::IndexingSessions { + indexed_sessions, + total_sessions, + } = event; + progress(RefreshProgress::IndexingSessions { + project_name: project_name.to_owned(), + indexed_sessions, + total_sessions, + }); + }, )?; progress(RefreshProgress::IndexFinished { project_name: project_name.to_owned(), diff --git a/crates/core/src/sync.rs b/crates/core/src/sync.rs index 02ada56..1fd5ef0 100644 --- a/crates/core/src/sync.rs +++ b/crates/core/src/sync.rs @@ -9,6 +9,7 @@ use darc_paths::{ normalize_project_path, normalized_known_paths, project_path_set, project_path_set_text_aliases, project_path_text_aliases, try_git_output, }; +pub(crate) use darc_sync::SyncProgress; use darc_sync::{ClaudeSource, CodexSource, SyncRequest}; use crate::{ @@ -98,6 +99,15 @@ pub fn prepare_sync(root: Option, options: SyncOptions) -> Result Result { + let mut progress = |_| {}; + execute_sync_with_progress(plan, &mut progress) +} + +/// Executes a prepared sync while reporting copied session progress. +pub(crate) fn execute_sync_with_progress( + plan: SyncPlan, + mut progress: impl FnMut(SyncProgress), +) -> Result { let SyncPlan { project_name, project_root, @@ -115,7 +125,7 @@ pub fn execute_sync(plan: SyncPlan) -> Result { config, } = writes; - let report = darc_sync::execute_sync(engine_plan)?; + let report = darc_sync::execute_sync_with_progress(engine_plan, &mut progress)?; if let Some(config) = &config { write_shared_config(&config_path, config)?; } diff --git a/crates/index/src/engine.rs b/crates/index/src/engine.rs index 67bf719..e2d96b5 100644 --- a/crates/index/src/engine.rs +++ b/crates/index/src/engine.rs @@ -35,6 +35,8 @@ use rusqlite::{Connection, OptionalExtension, Transaction, params}; use thiserror::Error; use walkdir::WalkDir; +const SESSION_PROGRESS_EMIT_INTERVAL: usize = 128; + /// Parses one Codex rollout file into user-visible turns. #[cfg(test)] pub(crate) fn parse_codex_rollout(path: &Path) -> Result { @@ -56,6 +58,15 @@ pub struct IndexReport { pub skipped_rollouts: Vec, } +/// Describes one observable indexing transition for progress UIs. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum IndexProgress { + IndexingSessions { + indexed_sessions: usize, + total_sessions: usize, + }, +} + /// Describes one archived rollout file that darc skipped during indexing. #[derive(Debug, Clone, PartialEq, Eq)] pub struct SkippedRollout { @@ -428,9 +439,24 @@ pub(crate) fn index_project_codex_turns_from( pub fn index_project_archived_sessions( request: &ProjectIndexRequest, providers: &[SourceKind], +) -> Result { + let mut progress = |_| {}; + index_project_archived_sessions_with_progress(request, providers, &mut progress) +} + +/// Indexes archived provider rollouts while reporting session progress. +pub fn index_project_archived_sessions_with_progress( + request: &ProjectIndexRequest, + providers: &[SourceKind], + mut progress: impl FnMut(IndexProgress), ) -> Result { let discovered_rollouts = discover_archived_rollouts(&request.sessions_root, providers)?; let mut connection = open_index_database(&request.index_db_path)?; + let total_sessions = discovered_rollouts.groups.len(); + progress(IndexProgress::IndexingSessions { + indexed_sessions: 0, + total_sessions, + }); let index_outcome = update_project_turns( &mut connection, @@ -438,6 +464,7 @@ pub fn index_project_archived_sessions( providers, &discovered_rollouts.groups, &discovered_rollouts.discovered_session_ids, + &mut progress, )?; let mut skipped_rollouts = discovered_rollouts.skipped_rollouts; skipped_rollouts.extend(index_outcome.skipped_rollouts); @@ -985,6 +1012,7 @@ fn update_project_turns( providers: &[SourceKind], archived_rollouts: &[ArchivedRolloutGroup], discovered_session_ids: &BTreeSet, + progress: &mut impl FnMut(IndexProgress), ) -> Result { let provider_set = providers.iter().copied().collect::>(); let mut transaction = connection @@ -1001,7 +1029,8 @@ fn update_project_turns( let mut sessions_succeeded = 0; let mut skipped_rollouts = Vec::new(); - for archived_group in archived_rollouts { + let total_sessions = archived_rollouts.len(); + for (index, archived_group) in archived_rollouts.iter().enumerate() { let group_outcome = update_archived_rollout_group( &mut transaction, project_id, @@ -1010,6 +1039,13 @@ fn update_project_turns( )?; sessions_succeeded += usize::from(group_outcome.session_succeeded); skipped_rollouts.extend(group_outcome.skipped_rollouts); + let indexed_sessions = index + 1; + if should_emit_session_progress(indexed_sessions, total_sessions) { + progress(IndexProgress::IndexingSessions { + indexed_sessions, + total_sessions, + }); + } } transaction @@ -1024,6 +1060,11 @@ fn update_project_turns( }) } +/// Returns whether one session-count progress event should be emitted. +fn should_emit_session_progress(current: usize, total: usize) -> bool { + current >= total || current.is_multiple_of(SESSION_PROGRESS_EMIT_INTERVAL) +} + /// Describes how one archived duplicate group affected the SQLite index. #[derive(Debug, Clone)] struct ArchivedRolloutGroupOutcome { diff --git a/crates/index/src/lib.rs b/crates/index/src/lib.rs index 73e653d..b0a1bbd 100644 --- a/crates/index/src/lib.rs +++ b/crates/index/src/lib.rs @@ -9,6 +9,7 @@ pub use darc_store::{ open_existing_index_database, open_index_database, open_index_database_read_only, policy, }; pub use engine::{ - IndexReport, ProjectIndexRequest, SkippedCodexRollout, SkippedRollout, + IndexProgress, IndexReport, ProjectIndexRequest, SkippedCodexRollout, SkippedRollout, index_project_archived_codex_turns, index_project_archived_sessions, + index_project_archived_sessions_with_progress, }; diff --git a/crates/sync/src/engine.rs b/crates/sync/src/engine.rs index e8bfb5a..73bd937 100644 --- a/crates/sync/src/engine.rs +++ b/crates/sync/src/engine.rs @@ -22,6 +22,8 @@ use crate::{ utils::{copy_file_atomically, file_snapshot, format_system_time_utc, write_json_atomically}, }; +const SESSION_PROGRESS_EMIT_INTERVAL: usize = 128; + /// Describes a prepared sync before any writes happen. #[derive(Debug, Clone)] pub struct SyncPlan { @@ -75,6 +77,15 @@ pub struct SyncReport { pub manifest_written: bool, } +/// Describes one observable sync transition for progress UIs. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SyncProgress { + CopyingSessions { + copied_sessions: usize, + total_sessions: usize, + }, +} + /// Plans a sync from explicit project and source inputs. pub fn prepare_sync(request: SyncRequest) -> Result { let manifest_path = request.sessions_root.join(".manifest.json"); @@ -164,6 +175,15 @@ pub fn prepare_sync(request: SyncRequest) -> Result { /// Executes a prepared sync by copying files and atomically updating metadata. pub fn execute_sync(plan: SyncPlan) -> Result { + let mut progress = |_| {}; + execute_sync_with_progress(plan, &mut progress) +} + +/// Executes a prepared sync while reporting copied session progress. +pub fn execute_sync_with_progress( + plan: SyncPlan, + mut progress: impl FnMut(SyncProgress), +) -> Result { let SyncPlan { project_name, project_root, @@ -184,7 +204,18 @@ pub fn execute_sync(plan: SyncPlan) -> Result { auxiliary_copies, } = writes; - for copy in session_copies.iter().chain(&auxiliary_copies) { + let total_sessions = session_copies.len(); + for (index, copy) in session_copies.iter().enumerate() { + copy_file_atomically(©.source_path, ©.destination_path)?; + let copied_sessions = index + 1; + if should_emit_session_progress(copied_sessions, total_sessions) { + progress(SyncProgress::CopyingSessions { + copied_sessions, + total_sessions, + }); + } + } + for copy in &auxiliary_copies { copy_file_atomically(©.source_path, ©.destination_path)?; } if manifest_written { @@ -206,6 +237,11 @@ pub fn execute_sync(plan: SyncPlan) -> Result { }) } +/// Returns whether one session-count progress event should be emitted. +fn should_emit_session_progress(current: usize, total: usize) -> bool { + current >= total || current.is_multiple_of(SESSION_PROGRESS_EMIT_INTERVAL) +} + /// Captures supported Claude discovery results. #[derive(Debug, Default)] pub(crate) struct ClaudeDiscovery { diff --git a/crates/sync/src/lib.rs b/crates/sync/src/lib.rs index da42ab0..b1f14a9 100644 --- a/crates/sync/src/lib.rs +++ b/crates/sync/src/lib.rs @@ -5,5 +5,7 @@ mod tests; mod types; pub(crate) mod utils; -pub use engine::{SyncPlan, SyncReport, execute_sync, prepare_sync}; +pub use engine::{ + SyncPlan, SyncProgress, SyncReport, execute_sync, execute_sync_with_progress, prepare_sync, +}; pub use types::{ClaudeSource, CodexSource, SourceKind, SyncRequest};