refactor(context_engine): remove retry logic and simplify error handling (#2407)

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

Author: tusharmath

Cargo.lock

@@ -17,6 +17,7 @@ strip = true
 [workspace.dependencies]
 anyhow = "1.0.101"
 async-recursion = "1.1.1"
+async-stream = "0.3"
 async-trait = "0.1.89"
 aws-config = { version = "1.8.13", features = ["behavior-version-latest"], default-features = false }
 aws-sdk-bedrockruntime = { version = "1.124.0", features = ["behavior-version-latest"], default-features = false }

Cargo.toml

@@ -45,6 +45,21 @@ pub trait FileReaderInfra: Send + Sync {
     /// Returns the file content as a UTF-8 string.
     async fn read_utf8(&self, path: &Path) -> anyhow::Result<String>;
 
+    /// Reads multiple files in batches and returns a stream of file batches.
+    ///
+    /// # Arguments
+    /// * `batch_size` - Number of files to include in each batch
+    /// * `paths` - Vector of file paths to read
+    ///
+    /// Returns a stream where each item is a vector of tuples containing
+    /// (file_path, file_content). Files within each batch are read concurrently
+    /// for better performance.
+    fn read_batch_utf8(
+        &self,
+        batch_size: usize,
+        paths: Vec<PathBuf>,
+    ) -> impl futures::Stream<Item = anyhow::Result<Vec<(PathBuf, String)>>> + Send;
+
     /// Reads the content of a file at the specified path.
     /// Returns the file content as raw bytes.
     async fn read(&self, path: &Path) -> anyhow::Result<Vec<u8>>;

crates/forge_app/src/infra.rs

@@ -75,6 +75,7 @@ impl Default for TestContext {
                 stdout_max_prefix_length: 256,
                 stdout_max_suffix_length: 256,
                 max_read_size: 4096,
+                max_file_read_batch_size: 50,
                 http: HttpConfig::default(),
                 max_file_size: 1024 * 1024 * 5,
                 max_search_result_bytes: 200,

crates/forge_app/src/orch_spec/orch_setup.rs

@@ -1,28 +1,25 @@
 use std::collections::{BTreeSet, HashMap};
 
-use forge_domain::{FileHash, FileNode, FileStatus, SyncProgress, SyncStatus};
+use forge_domain::{FileHash, FileStatus, SyncProgress, SyncStatus};
 
 /// Result of comparing local and server files
 ///
-/// This struct stores local and remote file information and provides methods
+/// This struct stores remote file information and provides methods
 /// to compute synchronization operations on-demand. It can derive file statuses
 /// and identify which files need to be uploaded, deleted, or modified.
 pub struct WorkspaceStatus {
-    /// Local files with their content and hashes
-    local_files: Vec<FileNode>,
     /// Remote file hashes from the server
     remote_files: Vec<FileHash>,
 }
 
 impl WorkspaceStatus {
-    /// Creates a sync plan from local files and remote file hashes.
+    /// Creates a sync plan from remote file hashes.
     ///
     /// # Arguments
     ///
-    /// * `local_files` - Vector of local files with their content and hashes
     /// * `remote_files` - Vector of remote file hashes from the server
-    pub fn new(local_files: Vec<FileNode>, remote_files: Vec<FileHash>) -> Self {
-        Self { local_files, remote_files }
+    pub fn new(remote_files: Vec<FileHash>) -> Self {
+        Self { remote_files }
     }
 
     /// Derives file sync statuses by comparing local and remote files.
@@ -34,12 +31,11 @@ impl WorkspaceStatus {
     /// - `Modified`: File exists in both but with different hashes
     /// - `New`: File exists only locally
     /// - `Deleted`: File exists only remotely
-    pub fn file_statuses(&self) -> Vec<FileStatus> {
+    pub fn file_statuses(&self, local_files: Vec<FileHash>) -> Vec<FileStatus> {
         // Build hash maps for efficient lookup
-        let local_hashes: HashMap<&str, &str> = self
-            .local_files
+        let local_hashes: HashMap<&str, &str> = local_files
             .iter()
-            .map(|f| (f.file_path.as_str(), f.hash.as_str()))
+            .map(|f| (f.path.as_str(), f.hash.as_str()))
             .collect();
         let remote_hashes: HashMap<&str, &str> = self
             .remote_files
@@ -78,38 +74,16 @@ impl WorkspaceStatus {
     ///
     /// A tuple of (files_to_delete, files_to_upload) where:
     /// - `files_to_delete`: Vector of file paths to delete from remote
-    /// - `files_to_upload`: Vector of files to upload to remote
-    pub fn get_operations(&self) -> (Vec<String>, Vec<forge_domain::FileRead>) {
-        let statuses = self.file_statuses();
+    /// - `files_to_upload`: Vector of file paths to upload to remote
+    pub fn get_operations(&self, local_files: Vec<FileHash>) -> (Vec<String>, Vec<String>) {
+        let statuses = self.file_statuses(local_files);
         let mut files_to_delete = Vec::new();
         let mut files_to_upload = Vec::new();
 
-        // Create a map for quick lookup of local files
-        let local_files_map: HashMap<&str, &FileNode> = self
-            .local_files
-            .iter()
-            .map(|f| (f.file_path.as_str(), f))
-            .collect();
-
         for status in statuses {
             match status.status {
-                SyncStatus::Modified => {
-                    // Note: Modified files already exist in the database and will be
-                    // automatically deleted by the backend.
-                    if let Some(file) = local_files_map.get(status.path.as_str()) {
-                        files_to_upload.push(forge_domain::FileRead::new(
-                            file.file_path.clone(),
-                            file.content.clone(),
-                        ));
-                    }
-                }
-                SyncStatus::New => {
-                    if let Some(file) = local_files_map.get(status.path.as_str()) {
-                        files_to_upload.push(forge_domain::FileRead::new(
-                            file.file_path.clone(),
-                            file.content.clone(),
-                        ));
-                    }
+                SyncStatus::Modified | SyncStatus::New => {
+                    files_to_upload.push(status.path);
                 }
                 SyncStatus::Deleted => {
                     files_to_delete.push(status.path);
@@ -123,7 +97,6 @@ impl WorkspaceStatus {
         (files_to_delete, files_to_upload)
     }
 }
-
 /// Tracks progress of sync operations
 pub struct SyncProgressCounter {
     total_files: usize,
@@ -162,30 +135,18 @@ mod tests {
     #[test]
     fn test_file_statuses() {
         let local = vec![
-            FileNode {
-                file_path: "a.rs".into(),
-                content: "content_a".into(),
-                hash: "hash_a".into(),
-            },
-            FileNode {
-                file_path: "b.rs".into(),
-                content: "modified_content".into(),
-                hash: "new_hash".into(),
-            },
-            FileNode {
-                file_path: "d.rs".into(),
-                content: "content_d".into(),
-                hash: "hash_d".into(),
-            },
+            FileHash { path: "a.rs".into(), hash: "hash_a".into() },
+            FileHash { path: "b.rs".into(), hash: "new_hash".into() },
+            FileHash { path: "d.rs".into(), hash: "hash_d".into() },
         ];
         let remote = vec![
             FileHash { path: "a.rs".into(), hash: "hash_a".into() },
             FileHash { path: "b.rs".into(), hash: "old_hash".into() },
             FileHash { path: "c.rs".into(), hash: "hash_c".into() },
         ];
 
-        let plan = WorkspaceStatus::new(local, remote);
-        let actual = plan.file_statuses();
+        let plan = WorkspaceStatus::new(remote);
+        let actual = plan.file_statuses(local);
 
         let expected = vec![
             forge_domain::FileStatus::new("a.rs".to_string(), forge_domain::SyncStatus::InSync),

crates/forge_app/src/workspace_status.rs

@@ -52,6 +52,9 @@ pub struct Environment {
     pub max_line_length: usize,
     /// Maximum number of lines to read from a file
     pub max_read_size: u64,
+    /// Maximum number of files that can be read in a single batch operation.
+    /// Controlled by FORGE_MAX_READ_BATCH_SIZE environment variable.
+    pub max_file_read_batch_size: usize,
     /// Http configuration
     pub http: HttpConfig,
     /// Maximum file size in bytes for operations
@@ -289,6 +292,7 @@ fn test_command_path() {
         stdout_max_line_length: 500,
         max_line_length: 2000,
         max_read_size: 2000,
+        max_file_read_batch_size: 50,
         http: HttpConfig::default(),
         max_file_size: 104857600,
         tool_timeout: 300,
@@ -329,6 +333,7 @@ fn test_command_cwd_path() {
         stdout_max_line_length: 500,
         max_line_length: 2000,
         max_read_size: 2000,
+        max_file_read_batch_size: 50,
         http: HttpConfig::default(),
         max_file_size: 104857600,
         tool_timeout: 300,
@@ -369,6 +374,7 @@ fn test_command_cwd_path_independent_from_command_path() {
         stdout_max_line_length: 500,
         max_line_length: 2000,
         max_read_size: 2000,
+        max_file_read_batch_size: 50,
         http: HttpConfig::default(),
         max_file_size: 104857600,
         tool_timeout: 300,

crates/forge_domain/src/env.rs

@@ -99,6 +99,9 @@ pub enum Error {
     #[error("Workspace already initialized with id: {0}")]
     WorkspaceAlreadyInitialized(WorkspaceId),
 
+    #[error("Failed to sync {count} file(s)")]
+    SyncFailed { count: usize },
+
     #[error("No default provider set.")]
     NoDefaultProvider,
 
@@ -149,6 +152,10 @@ impl Error {
     pub fn no_default_model(provider: ProviderId) -> Self {
         Self::NoDefaultModel(provider)
     }
+
+    pub fn sync_failed(count: usize) -> Self {
+        Self::SyncFailed { count }
+    }
 }
 
 #[cfg(test)]

crates/forge_domain/src/error.rs

@@ -42,6 +42,12 @@ pub struct FileHash {
     pub hash: String,
 }
 
+impl From<super::node::FileNode> for FileHash {
+    fn from(node: super::node::FileNode) -> Self {
+        Self { path: node.file_path, hash: node.hash }
+    }
+}
+
 /// Status of a file in relation to the server
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, PartialOrd, Ord)]
 pub enum SyncStatus {

crates/forge_domain/src/file.rs

@@ -6,7 +6,7 @@ use url::Url;
 use crate::{
     AnyProvider, AppConfig, AuthCredential, ChatCompletionMessage, Context, Conversation,
     ConversationId, MigrationResult, Model, ModelId, Provider, ProviderId, ProviderTemplate,
-    ResultStream, SearchMatch, Skill, Snapshot, UserId, Workspace, WorkspaceAuth, WorkspaceId,
+    ResultStream, SearchMatch, Skill, Snapshot, WorkspaceAuth, WorkspaceId,
 };
 
 /// Repository for managing file snapshots
@@ -116,40 +116,6 @@ pub trait ProviderRepository: Send + Sync {
     async fn migrate_env_credentials(&self) -> anyhow::Result<Option<MigrationResult>>;
 }
 
-/// Repository for managing workspace metadata in local database
-#[async_trait::async_trait]
-pub trait WorkspaceRepository: Send + Sync {
-    /// Save or update a workspace
-    async fn upsert(
-        &self,
-        workspace_id: &WorkspaceId,
-        user_id: &UserId,
-        path: &std::path::Path,
-    ) -> anyhow::Result<()>;
-
-    /// Find all workspaces for a user
-    ///
-    /// Returns all workspaces belonging to the specified user.
-    /// Path matching and selection logic should be handled in the service
-    /// layer.
-    ///
-    /// # Arguments
-    /// * `user_id` - Only return workspaces belonging to this user
-    ///
-    /// # Returns
-    /// A vector of all workspaces for the user (may be empty)
-    ///
-    /// # Errors
-    /// Returns an error if there's a database error
-    async fn list(&self) -> anyhow::Result<Vec<Workspace>>;
-
-    /// Get user ID from any workspace, or None if no workspaces exist
-    async fn get_user_id(&self) -> anyhow::Result<Option<UserId>>;
-
-    /// Delete workspace from local database
-    async fn delete(&self, workspace_id: &WorkspaceId) -> anyhow::Result<()>;
-}
-
 /// Repository for managing workspace indexing and search operations
 #[async_trait::async_trait]
 pub trait WorkspaceIndexRepository: Send + Sync {

crates/forge_domain/src/repo.rs

@@ -2,8 +2,6 @@ use derive_more::Display;
 use serde::{Deserialize, Serialize};
 use uuid::Uuid;
 
-use crate::UserId;
-
 /// Workspace identifier (UUID) from workspace server.
 ///
 /// Generated locally and sent to server during CreateWorkspace.
@@ -30,13 +28,3 @@ impl WorkspaceId {
         self.0
     }
 }
-
-/// Domain entity for workspace
-#[derive(Debug, Clone)]
-pub struct Workspace {
-    pub workspace_id: WorkspaceId,
-    pub user_id: UserId,
-    pub path: std::path::PathBuf,
-    pub created_at: chrono::DateTime<chrono::Utc>,
-    pub updated_at: Option<chrono::DateTime<chrono::Utc>>,
-}