Merge PR #254: fix: batch fixes for issues #2329-#2352

Author: factorydroid

Cargo.lock

@@ -62,6 +62,7 @@ url = { workspace = true }
 bytes = { workspace = true }
 base64 = { workspace = true }
 dirs = "5"
+fs2 = "0.4"  # File locking for concurrent access
 
 # Authentication
 jsonwebtoken = "9"

cortex-app-server/Cargo.toml

@@ -316,6 +316,45 @@ pub async fn error_handling_middleware(request: Request, next: Next) -> Response
     response
 }
 
+/// Content negotiation middleware.
+///
+/// This middleware validates the Accept header and returns 406 Not Acceptable
+/// if the client requests an unsupported content type. The API only supports
+/// JSON responses.
+///
+/// Supported content types:
+/// - `application/json`
+/// - `*/*` (wildcard)
+/// - No Accept header (defaults to JSON)
+pub async fn content_negotiation_middleware(
+    request: Request,
+    next: Next,
+) -> Result<Response, StatusCode> {
+    // Get Accept header
+    if let Some(accept) = request.headers().get(header::ACCEPT) {
+        if let Ok(accept_str) = accept.to_str() {
+            // Parse Accept header and check for supported types
+            let supported = accept_str.split(',').any(|media_type| {
+                let media_type = media_type.split(';').next().unwrap_or("").trim();
+                media_type == "application/json"
+                    || media_type == "application/*"
+                    || media_type == "*/*"
+                    || media_type.is_empty()
+            });
+
+            if !supported {
+                warn!(
+                    "Unsupported Accept header: {}. Only application/json is supported.",
+                    accept_str
+                );
+                return Err(StatusCode::NOT_ACCEPTABLE);
+            }
+        }
+    }
+
+    Ok(next.run(request).await)
+}
+
 /// Request context extracted from middleware.
 #[derive(Debug, Clone)]
 pub struct RequestContext {

cortex-app-server/src/middleware.rs

@@ -1,11 +1,14 @@
 //! Session persistence storage for cortex-app-server.
 //!
 //! Saves sessions and message history to disk so they survive server restarts.
+//! Uses file locking to prevent corruption when multiple server instances share
+//! the same storage directory.
 
 use std::fs;
 use std::io::{BufRead, BufReader, BufWriter, Write};
 use std::path::{Path, PathBuf};
 
+use fs2::FileExt;
 use serde::{Deserialize, Serialize};
 use tracing::{debug, error, info, warn};
 
@@ -78,24 +81,45 @@ impl SessionStorage {
         Self::new(base_dir)
     }
 
-    /// Save a session to disk.
+    /// Save a session to disk with exclusive file locking.
+    ///
+    /// Uses file locking to prevent concurrent write corruption when multiple
+    /// server instances share the same storage directory.
     pub fn save_session(&self, session: &StoredSession) -> std::io::Result<()> {
         let path = self.session_path(&session.id);
         let file = fs::File::create(&path)?;
-        let writer = BufWriter::new(file);
-        serde_json::to_writer_pretty(writer, session)
-            .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))?;
+
+        // Acquire exclusive lock for writing
+        file.lock_exclusive()?;
+
+        let writer = BufWriter::new(&file);
+        let result = serde_json::to_writer_pretty(writer, session)
+            .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e));
+
+        // Lock is automatically released when file is dropped
+        file.unlock()?;
+
+        result?;
         debug!("Saved session {} to {:?}", session.id, path);
         Ok(())
     }
 
-    /// Load a session from disk.
+    /// Load a session from disk with shared file locking.
+    ///
+    /// Uses shared locking to allow concurrent reads while preventing
+    /// reads during writes.
     pub fn load_session(&self, id: &str) -> std::io::Result<StoredSession> {
         let path = self.session_path(id);
         let file = fs::File::open(&path)?;
-        let reader = BufReader::new(file);
+
+        // Acquire shared lock for reading
+        file.lock_shared()?;
+
+        let reader = BufReader::new(&file);
         let session: StoredSession = serde_json::from_reader(reader)
             .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
+
+        file.unlock()?;
         Ok(session)
     }
 
@@ -142,22 +166,31 @@ impl SessionStorage {
 
     /// Append a message to session history (JSONL format).
     ///
-    /// This function ensures data durability by calling sync_all() (fsync)
-    /// after writing to prevent data loss on crash or forceful termination.
+    /// This function uses file locking to prevent concurrent write corruption
+    /// and ensures data durability by calling sync_all() (fsync) after writing.
     pub fn append_message(&self, session_id: &str, message: &StoredMessage) -> std::io::Result<()> {
         let path = self.history_path(session_id);
-        let mut file = fs::OpenOptions::new()
+        let file = fs::OpenOptions::new()
             .create(true)
             .append(true)
             .open(&path)?;
 
+        // Acquire exclusive lock for writing
+        file.lock_exclusive()?;
+
         let json = serde_json::to_string(message)
             .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))?;
-        writeln!(file, "{}", json)?;
+
+        // Write using a mutable reference to the locked file
+        use std::io::Write;
+        let mut writer = &file;
+        writeln!(writer, "{}", json)?;
 
         // Ensure data is durably written to disk (fsync) to prevent data loss on crash
         file.sync_all()?;
 
+        file.unlock()?;
+
         debug!("Appended message to session {} history", session_id);
         Ok(())
     }

cortex-app-server/src/storage.rs

@@ -31,6 +31,9 @@ pub enum AgentSubcommand {
     /// Create a new agent interactively.
     Create(CreateArgs),
 
+    /// Edit an existing agent in your default editor.
+    Edit(EditArgs),
+
     /// Remove a user-defined agent.
     Remove(RemoveArgs),
 }
@@ -113,6 +116,17 @@ pub struct CreateArgs {
     pub model: String,
 }
 
+/// Arguments for edit command.
+#[derive(Debug, Parser)]
+pub struct EditArgs {
+    /// Name of the agent to edit.
+    pub name: String,
+
+    /// Editor to use (defaults to $EDITOR or $VISUAL).
+    #[arg(short, long)]
+    pub editor: Option<String>,
+}
+
 /// Arguments for remove command.
 #[derive(Debug, Parser)]
 pub struct RemoveArgs {
@@ -292,6 +306,7 @@ impl AgentCli {
             AgentSubcommand::List(args) => run_list(args).await,
             AgentSubcommand::Show(args) => run_show(args).await,
             AgentSubcommand::Create(args) => run_create(args).await,
+            AgentSubcommand::Edit(args) => run_edit(args).await,
             AgentSubcommand::Remove(args) => run_remove(args).await,
         }
     }
@@ -1291,6 +1306,121 @@ mode: {mode}
     Ok(())
 }
 
+/// Edit agent command.
+///
+/// Opens the agent file in the user's default editor, then validates the file
+/// after editing. If validation fails, offers to re-open the editor to fix issues.
+async fn run_edit(args: EditArgs) -> Result<()> {
+    let agents = load_all_agents()?;
+
+    let agent = agents
+        .iter()
+        .find(|a| a.name == args.name)
+        .ok_or_else(|| anyhow::anyhow!("Agent '{}' not found", args.name))?;
+
+    if agent.native {
+        bail!(
+            "Cannot edit built-in agent '{}'.\n\n\
+            Built-in agents are part of the Cortex core and cannot be modified.\n\
+            To customize this agent, create a copy:\n\
+            cortex agent create my-{}",
+            args.name,
+            args.name
+        );
+    }
+
+    let path = agent
+        .path
+        .as_ref()
+        .ok_or_else(|| anyhow::anyhow!("Agent '{}' has no file path", args.name))?;
+
+    // Determine the editor to use
+    let editor = args
+        .editor
+        .or_else(|| std::env::var("VISUAL").ok())
+        .or_else(|| std::env::var("EDITOR").ok())
+        .unwrap_or_else(|| {
+            if cfg!(windows) {
+                "notepad".to_string()
+            } else {
+                "vi".to_string()
+            }
+        });
+
+    // Make a backup of the original file
+    let backup_content = std::fs::read_to_string(path)
+        .with_context(|| format!("Failed to read agent file: {}", path.display()))?;
+
+    loop {
+        // Open the editor
+        println!("Opening {} in {}...", path.display(), editor);
+        let status = std::process::Command::new(&editor)
+            .arg(path)
+            .status()
+            .with_context(|| format!("Failed to launch editor: {}", editor))?;
+
+        if !status.success() {
+            bail!("Editor exited with error");
+        }
+
+        // Read and validate the edited file
+        let content = std::fs::read_to_string(path)
+            .with_context(|| format!("Failed to read edited file: {}", path.display()))?;
+
+        // Try to parse the frontmatter to validate
+        match parse_frontmatter(&content) {
+            Ok((frontmatter, _body)) => {
+                // Validate required fields
+                if frontmatter.name.trim().is_empty() {
+                    eprintln!("\nError: Agent name cannot be empty.");
+                } else if !frontmatter
+                    .name
+                    .chars()
+                    .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_')
+                {
+                    eprintln!(
+                        "\nError: Agent name must contain only alphanumeric characters, hyphens, and underscores."
+                    );
+                } else {
+                    // Validation passed
+                    println!("\nAgent '{}' updated successfully!", frontmatter.name);
+                    return Ok(());
+                }
+            }
+            Err(e) => {
+                eprintln!("\nError: Invalid agent configuration: {}", e);
+            }
+        }
+
+        // Validation failed - offer to re-edit or rollback
+        print!(
+            "Would you like to (e)dit again, (r)ollback to original, or (k)eep invalid file? [e/r/k]: "
+        );
+        io::stdout().flush()?;
+
+        let mut input = String::new();
+        io::stdin().lock().read_line(&mut input)?;
+
+        match input.trim().to_lowercase().as_str() {
+            "r" | "rollback" => {
+                // Restore the backup
+                std::fs::write(path, &backup_content)
+                    .with_context(|| format!("Failed to restore backup: {}", path.display()))?;
+                println!("Rolled back to original version.");
+                return Ok(());
+            }
+            "k" | "keep" => {
+                eprintln!("Warning: Keeping invalid configuration. The agent may fail to load.");
+                return Ok(());
+            }
+            _ => {
+                // Default: re-edit
+                continue;
+            }
+        }
+    }
+}
+
 /// Remove agent command.
 async fn run_remove(args: RemoveArgs) -> Result<()> {
     let agents = load_all_agents()?;