fix(cortex-engine): handle SIGPIPE by detecting closed event channel

Fixes bounty issue #1525

When cortex run is piped to a command that closes the pipe (e.g., head -n 1),
the session event channel could return a Closed error. Previously this was ignored,
causing the session to keep running. Now we detect TrySendError::Closed and properly
shut down the session.

Author: factorydroid

cortex-engine/src/session.rs

@@ -30,7 +30,7 @@ use crate::error::{CortexError, Result};
 use crate::rollout::reader::{RolloutItem, get_events, get_session_meta};
 use crate::rollout::recorder::SessionMeta;
 use crate::rollout::{RolloutRecorder, SESSIONS_SUBDIR, get_rollout_path, read_rollout};
-use crate::skills::{SkillsManager, render_skills_section, try_get_skills_manager};
+use crate::skills::{render_skills_section, try_get_skills_manager};
 use crate::summarization::SummarizationStrategy;
 use crate::tools::context::ToolOutputChunk;
 use crate::tools::{ToolContext, ToolRouter};
@@ -205,7 +205,10 @@ impl Session {
         // Initialize with system prompt (including skills if available)
         let skills_section = load_skills_section(&config.cwd);
         let mut messages = Vec::new();
-        messages.push(Message::system(build_system_prompt_with_skills(&config, skills_section.as_deref())));
+        messages.push(Message::system(build_system_prompt_with_skills(
+            &config,
+            skills_section.as_deref(),
+        )));
 
         // Initialize snapshot manager
         let snapshot_dir = config
@@ -1355,8 +1358,21 @@ impl Session {
             let _ = recorder.record_event(&event);
         }
 
-        // Non-blocking send - never wait
-        let _ = self.event_tx.try_send(event);
+        // Non-blocking send
+        // If the channel is closed (receiver dropped), we should stop the session
+        match self.event_tx.try_send(event) {
+            Ok(_) => {}
+            Err(async_channel::TrySendError::Full(_)) => {
+                // Channel full - dropping event (shouldn't happen with unbounded)
+                tracing::warn!("Event channel full, dropping event");
+            }
+            Err(async_channel::TrySendError::Closed(_)) => {
+                // Channel closed - receiver is gone (e.g. pipe broken)
+                tracing::info!("Event channel closed, stopping session");
+                self.running = false;
+                self.cancelled.store(true, Ordering::SeqCst);
+            }
+        }
     }
 
     /// Resume a session from a rollout file.
@@ -1794,12 +1810,12 @@ fn build_system_prompt_with_skills(config: &Config, skills_section: Option<&str>
     }
 
     // Add skills section if provided
-    if let Some(skills) = skills_section {
-        if !skills.is_empty() {
-            additional.push_str("\n## Available Skills\n");
-            additional.push_str(skills);
-            additional.push('\n');
-        }
+    if let Some(skills) = skills_section
+        && !skills.is_empty()
+    {
+        additional.push_str("\n## Available Skills\n");
+        additional.push_str(skills);
+        additional.push('\n');
     }
 
     prompt = prompt.replace("{{ADDITIONAL_CONTEXT}}", &additional);
@@ -1897,14 +1913,14 @@ fn get_system_info() -> String {
 fn load_skills_section(cwd: &PathBuf) -> Option<String> {
     // Try to get the global skills manager
     let manager = try_get_skills_manager()?;
-    
+
     // Load skills for current working directory (synchronously)
     let outcome = manager.skills_for_cwd_sync(cwd, false);
-    
+
     if outcome.skills.is_empty() {
         return None;
     }
-    
+
     // Render skills section for system prompt
     render_skills_section(&outcome.skills)
 }

cortex-engine/src/tasks/snapshot.rs

@@ -292,7 +292,7 @@ impl SnapshotManager {
             match state {
                 FileState::Exists {
                     content,
-                    permissions: _,
+                    permissions,
                     ..
                 } => {
                     // Create parent directories