fix(sync): use env-based batch size for consistent file-read concurrency (#2613)

Co-authored-by: Tushar Mathur <tusharmath@gmail.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

Author: 3 people

crates/forge_api/src/api.rs

@@ -207,7 +207,6 @@ pub trait API: Sync + Send {
     async fn sync_workspace(
         &self,
         path: PathBuf,
-        batch_size: usize,
     ) -> Result<MpscStream<Result<forge_domain::SyncProgress>>>;
 
     /// Query the indexed workspace

crates/forge_api/src/forge_api.rs

@@ -360,9 +360,8 @@ impl<
     async fn sync_workspace(
         &self,
         path: PathBuf,
-        batch_size: usize,
     ) -> Result<MpscStream<Result<forge_domain::SyncProgress>>> {
-        self.services.sync_workspace(path, batch_size).await
+        self.services.sync_workspace(path).await
     }
 
     async fn query_workspace(

crates/forge_app/src/services.rs

@@ -305,7 +305,6 @@ pub trait WorkspaceService: Send + Sync {
     async fn sync_workspace(
         &self,
         path: PathBuf,
-        batch_size: usize,
     ) -> anyhow::Result<forge_stream::MpscStream<anyhow::Result<SyncProgress>>>;
 
     /// Query the indexed workspace with semantic search
@@ -1110,11 +1109,8 @@ impl<I: Services> WorkspaceService for I {
     async fn sync_workspace(
         &self,
         path: PathBuf,
-        batch_size: usize,
     ) -> anyhow::Result<forge_stream::MpscStream<anyhow::Result<SyncProgress>>> {
-        self.workspace_service()
-            .sync_workspace(path, batch_size)
-            .await
+        self.workspace_service().sync_workspace(path).await
     }
 
     async fn query_workspace(

crates/forge_main/src/built_in_commands.json

@@ -103,6 +103,10 @@
     "command": "sync-info",
     "description": "Show workspace information with sync details"
   },
+  {
+    "command": "sync-init",
+    "description": "Initialize a new workspace without syncing files"
+  },
   {
     "command": "clone",
     "description": "Clone and manage conversation context"

crates/forge_main/src/cli.rs

@@ -242,9 +242,10 @@ pub enum WorkspaceCommand {
         #[arg(default_value = ".")]
         path: PathBuf,
 
-        /// Number of files to process concurrently
-        #[arg(long, default_value = "100")]
-        batch_size: usize,
+        /// Automatically initialize the workspace before syncing if it has not
+        /// been initialized yet.
+        #[arg(long)]
+        init: bool,
     },
     /// List all workspaces.
     List {

crates/forge_main/src/ui.rs

@@ -600,8 +600,8 @@ impl<A: API + ConsoleWriter + 'static, F: Fn() -> A + Send + Sync> UI<A, F> {
             }
             TopLevelCommand::Workspace(index_group) => {
                 match index_group.command {
-                    crate::cli::WorkspaceCommand::Sync { path, batch_size } => {
-                        self.on_index(path, batch_size).await?;
+                    crate::cli::WorkspaceCommand::Sync { path, init } => {
+                        self.on_index(path, init).await?;
                     }
                     crate::cli::WorkspaceCommand::List { porcelain } => {
                         self.on_list_workspaces(porcelain).await?;
@@ -1980,8 +1980,7 @@ impl<A: API + ConsoleWriter + 'static, F: Fn() -> A + Send + Sync> UI<A, F> {
             }
             SlashCommand::Index => {
                 let working_dir = self.state.cwd.clone();
-                // Use default batch size of 100 for slash command
-                self.on_index(working_dir, 100).await?;
+                self.on_index(working_dir, false).await?;
             }
             SlashCommand::AgentSwitch(agent_id) => {
                 // Validate that the agent exists by checking against loaded agents
@@ -3634,11 +3633,7 @@ impl<A: API + ConsoleWriter + 'static, F: Fn() -> A + Send + Sync> UI<A, F> {
         Ok(())
     }
 
-    async fn on_index(
-        &mut self,
-        path: std::path::PathBuf,
-        batch_size: usize,
-    ) -> anyhow::Result<()> {
+    async fn on_index(&mut self, path: std::path::PathBuf, init: bool) -> anyhow::Result<()> {
         use forge_domain::SyncProgress;
         use forge_spinner::ProgressBarManager;
 
@@ -3647,7 +3642,17 @@ impl<A: API + ConsoleWriter + 'static, F: Fn() -> A + Send + Sync> UI<A, F> {
             self.init_forge_services().await?;
         }
 
-        let mut stream = self.api.sync_workspace(path.clone(), batch_size).await?;
+        // When init is set, check if the workspace is already initialized
+        // via get_workspace_info before calling init, so we only initialize
+        // when a workspace does not yet exist for the given path.
+        if init {
+            let workspace_info = self.api.get_workspace_info(path.clone()).await?;
+            if workspace_info.is_none() {
+                self.on_workspace_init(path.clone()).await?;
+            }
+        }
+
+        let mut stream = self.api.sync_workspace(path.clone()).await?;
         let mut progress_bar = ProgressBarManager::default();
 
         while let Some(event) = stream.next().await {
@@ -3998,19 +4003,20 @@ impl<A: API + ConsoleWriter + 'static, F: Fn() -> A + Send + Sync> UI<A, F> {
 
     /// Initialize workspace for a directory without syncing files
     async fn on_workspace_init(&mut self, path: std::path::PathBuf) -> anyhow::Result<()> {
+        // Check if auth already exists and create if needed
+        if !self.api.is_authenticated().await? {
+            self.init_forge_services().await?;
+        }
+
         self.spinner.start(Some("Initializing workspace"))?;
 
         let workspace_id = self.api.init_workspace(path.clone()).await?;
 
         self.spinner.stop(None)?;
 
-        // Resolve and display the path
-        let canonical_path = path.canonicalize().unwrap_or_else(|_| path.clone());
-
         self.writeln_title(
             TitleFormat::info("Workspace initialized successfully")
-                .sub_title(format!("Path: {}", canonical_path.display()))
-                .sub_title(format!("Workspace ID: {}", workspace_id)),
+                .sub_title(format!("{}", workspace_id)),
         )?;
 
         Ok(())